]> git.saurik.com Git - wxWidgets.git/blob - samples/sockets/baseserver.cpp
projects / wxWidgets.git / blob
? search:


67e5ebc495d69154ab27628b2ccad840a3a349b2
[wxWidgets.git] / samples / sockets / baseserver.cpp
1 /////////////////////////////////////////////////////////////////////////////
2 // Name: samples/sockbase/client.cpp
3 // Purpose: Sockets sample for wxBase
4 // Author: Lukasz Michalski
5 // Modified by:
6 // Created: 27.06.2005
7 // Copyright: (c) 2005 Lukasz Michalski <lmichalski@user.sourceforge.net>
8 // Licence: wxWindows licence
9 /////////////////////////////////////////////////////////////////////////////
10
11 // ============================================================================
12 // declarations
13 // ============================================================================
14
15 // ----------------------------------------------------------------------------
16 // headers
17 // ----------------------------------------------------------------------------
18
19 #include "wx/wx.h"
20 #include "wx/socket.h"
21 #include "wx/event.h"
22 #include "wx/list.h"
23 #include "wx/cmdline.h"
24 #include "wx/datetime.h"
25 #include "wx/timer.h"
26 #include "wx/thread.h"
27
28 const char *GetSocketErrorMsg(int pSockError)
29 {
30 switch(pSockError)
31 {
32 case wxSOCKET_NOERROR:
33 return "wxSOCKET_NOERROR";
34
35 case wxSOCKET_INVOP:
36 return "wxSOCKET_INVOP";
37
38 case wxSOCKET_IOERR:
39 return "wxSOCKET_IOERR";
40
41 case wxSOCKET_INVADDR:
42 return "wxSOCKET_INVADDR";
43
44 case wxSOCKET_NOHOST:
45 return "wxSOCKET_NOHOST";
46
47 case wxSOCKET_INVPORT:
48 return "wxSOCKET_INVPORT";
49
50 case wxSOCKET_WOULDBLOCK:
51 return "wxSOCKET_WOULDBLOCK";
52
53 case wxSOCKET_TIMEDOUT:
54 return "wxSOCKET_TIMEDOUT";
55
56 case wxSOCKET_MEMERR:
57 return "wxSOCKET_MEMERR";
58
59 default:
60 return "Unknown";
61 }
62 }
63
64 //event sent by workers to server class
65 //after client is served
66 const wxEventType wxEVT_WORKER = wxNewEventType();
67 #define EVT_WORKER(func) DECLARE_EVENT_TABLE_ENTRY( wxEVT_WORKER, -1, -1, (wxObjectEventFunction) (wxEventFunction) (WorkerEventFunction) & func, (wxObject *) NULL ),
68
69 class WorkerEvent : public wxEvent
70 {
71 public:
72 WorkerEvent(void* pSender)
73 {
74 SetId(-1);
75 SetEventType(wxEVT_WORKER);
76 m_sender = pSender;
77 m_exit = false;
78 m_workerFailed = false;
79 }
80
81 virtual wxEvent* Clone() const
82 {
83 return new WorkerEvent(*this);
84 }
85
86 void* m_sender;
87 bool m_exit;
88 bool m_workerFailed;
89 };
90
91 typedef void (wxEvtHandler::*WorkerEventFunction)(WorkerEvent&);
92
93 class ThreadWorker;
94 class EventWorker;
95
96 WX_DECLARE_LIST(ThreadWorker, TList);
97 WX_DECLARE_LIST(EventWorker, EList);
98
99 //main server class contains listening socket
100 //and list of two type worker classes that serve clients
101 class Server : public wxApp
102 {
103 DECLARE_EVENT_TABLE()
104 public:
105 Server() : m_maxConnections(-1) {}
106 ~Server() {}
107 private:
108 enum WorkMode
109 {
110 MIXED,
111 THREADS,
112 EVENTS
113 };
114
115 virtual bool OnInit();
116 virtual int OnExit();
117
118 void OnInitCmdLine(wxCmdLineParser& pParser);
119 bool OnCmdLineParsed(wxCmdLineParser& pParser);
120
121 void OnSocketEvent(wxSocketEvent& pEvent);
122 void OnWorkerEvent(WorkerEvent& pEvent);
123 void OnTimerEvent(wxTimerEvent& pEvent);
124 void DumpStatistics();
125
126 TList m_threadWorkers;
127 EList m_eventWorkers;
128 WorkMode m_workMode;
129 wxSocketServer* m_listeningSocket;
130
131 // statistics
132 unsigned m_threadWorkersCreated;
133 unsigned m_threadWorkersDone;
134 unsigned m_threadWorkersFailed;
135 unsigned m_maxThreadWorkers;
136
137 unsigned m_eventWorkersCreated;
138 unsigned m_eventWorkersDone;
139 unsigned m_eventWorkersFailed;
140 unsigned m_maxEventWorkers;
141
142 long int m_maxConnections;
143
144 unsigned short m_port;
145
146 wxTimer mTimer;
147 };
148
149 DECLARE_APP(Server);
150
151 // just some common things shared between ThreadWorker and EventWorker
152 class WorkerBase
153 {
154 protected:
155 // outputs log message with IP and TCP port number prepended
156 void LogWorker(const wxString& msg, wxLogLevel level = wxLOG_Info)
157 {
158 wxLogGeneric(level,
159 "%s:%d %s", m_peer.IPAddress(), m_peer.Service(), msg);
160 }
161
162 wxIPV4address m_peer;
163 };
164
165 //thread based worker reads signature and all data first from connected client
166 //and resends data to client after reading
167 class ThreadWorker : public wxThread, private WorkerBase
168 {
169 public:
170 ThreadWorker(wxSocketBase* pSocket);
171 virtual ExitCode Entry();
172
173 private:
174 wxSocketBase* m_socket;
175 };
176
177 //event based worker reads signature and creates buffer for incoming data.
178 //When part of data arrives this worker resends it as soon as possible.
179 class EventWorker : public wxEvtHandler, private WorkerBase
180 {
181 public:
182 EventWorker(wxSocketBase* pSock);
183 virtual ~EventWorker();
184
185 private:
186 wxSocketBase* m_socket;
187
188 unsigned char m_signature[2];
189 char* m_inbuf;
190 int m_infill;
191 int m_size;
192 char* m_outbuf;
193 int m_outfill;
194 int m_written;
195
196 void OnSocketEvent(wxSocketEvent& pEvent);
197 void DoWrite();
198 void DoRead();
199
200 DECLARE_EVENT_TABLE()
201 };
202
203 /******************* Implementation ******************/
204 IMPLEMENT_APP_CONSOLE(Server)
205
206 #include <wx/listimpl.cpp>
207 WX_DEFINE_LIST(TList);
208 WX_DEFINE_LIST(EList);
209
210
211 void
212 Server::OnInitCmdLine(wxCmdLineParser& pParser)
213 {
214 wxApp::OnInitCmdLine(pParser);
215 pParser.AddSwitch("t","threads","Use thread based workers only");
216 pParser.AddSwitch("e","events","Use event based workers only");
217 pParser.AddOption("m","max","Exit after <n> connections",
218 wxCMD_LINE_VAL_NUMBER);
219 pParser.AddOption("p","port","listen on given port (default 3000)",
220 wxCMD_LINE_VAL_NUMBER);
221 }
222
223 void
224 Server::DumpStatistics()
225 {
226 wxString mode;
227 switch(m_workMode)
228 {
229 case EVENTS:
230 mode = "Event based workers";
231 break;
232 case THREADS:
233 mode = "Thread based workers";
234 break;
235 case MIXED:
236 mode = "Event and thread based workers";
237 break;
238 }
239 wxLogMessage("Server mode: %s",mode);
240 wxLogMessage("\t\t\t\tThreads\tEvents\tTotal");
241 wxLogMessage("Workers created:\t\t%d\t%d\t%d",
242 m_threadWorkersCreated,
243 m_eventWorkersCreated,
244 m_threadWorkersCreated + m_eventWorkersCreated);
245 wxLogMessage("Max concurrent workers:\t%d\t%d\t%d",
246 m_maxThreadWorkers,
247 m_maxEventWorkers,
248 m_maxThreadWorkers + m_maxEventWorkers);
249 wxLogMessage("Workers failed:\t\t%d\t%d\t%d",
250 m_threadWorkersFailed,
251 m_eventWorkersFailed,
252 m_threadWorkersFailed + m_eventWorkersFailed);
253 wxLogMessage("Workers done:\t\t%d\t%d\t%d",
254 m_threadWorkersDone,
255 m_eventWorkersDone,
256 m_threadWorkersDone + m_eventWorkersDone);
257
258 if ((int)(m_threadWorkersDone+m_eventWorkersDone) == m_maxConnections)
259 {
260 wxLogMessage("%ld connection(s) served, exiting",m_maxConnections);
261 ExitMainLoop();
262 }
263 }
264
265
266 bool
267 Server::OnCmdLineParsed(wxCmdLineParser& pParser)
268 {
269 if (pParser.Found("verbose"))
270 {
271 wxLog::AddTraceMask("wxSocket");
272 wxLog::AddTraceMask("epolldispatcher");
273 wxLog::AddTraceMask("selectdispatcher");
274 wxLog::AddTraceMask("thread");
275 wxLog::AddTraceMask("events");
276 wxLog::AddTraceMask("timer");
277 }
278
279 if (pParser.Found("m",&m_maxConnections))
280 {
281 wxLogMessage("%ld connection(s) to exit",m_maxConnections);
282 }
283
284 long port;
285 if (pParser.Found("p", &port))
286 {
287 if ( port <= 0 || port > USHRT_MAX )
288 {
289 wxLogError("Invalid port number %ld, must be in 0..%u range.",
290 port, USHRT_MAX);
291 return false;
292 }
293
294 m_port = static_cast<unsigned short>(port);
295 wxLogMessage("Will listen on port %u", m_port);
296 }
297
298 if (pParser.Found("t"))
299 m_workMode = THREADS;
300 else if (pParser.Found("e"))
301 m_workMode = EVENTS;
302 else
303 m_workMode = MIXED;
304
305 return wxApp::OnCmdLineParsed(pParser);
306 }
307
308 bool Server::OnInit()
309 {
310 wxLog* logger = new wxLogStderr();
311 wxLog::SetActiveTarget(logger);
312
313 m_port = 3000;
314
315 //send interesting things to console
316 if (!wxApp::OnInit())
317 return false;
318
319 //setup listening socket
320 wxIPV4address la;
321 la.Service(m_port);
322 m_listeningSocket = new wxSocketServer(la,wxSOCKET_NOWAIT|wxSOCKET_REUSEADDR);
323 m_listeningSocket->SetEventHandler(*this);
324 m_listeningSocket->SetNotify(wxSOCKET_CONNECTION_FLAG);
325 m_listeningSocket->Notify(true);
326 if (!m_listeningSocket->IsOk())
327 {
328 wxLogError("Cannot bind listening socket");
329 return false;
330 }
331
332 m_threadWorkersCreated = 0;
333 m_threadWorkersDone = 0;
334 m_threadWorkersFailed = 0;
335 m_maxThreadWorkers = 0;
336
337 m_eventWorkersCreated = 0;
338 m_eventWorkersDone = 0;
339 m_eventWorkersFailed = 0;
340 m_maxEventWorkers = 0;
341
342 wxLogMessage("Server listening at port %u, waiting for connections", m_port);
343 return true;
344 }
345
346
347 int Server::OnExit()
348 {
349 for ( TList::compatibility_iterator it = m_threadWorkers.GetFirst();
350 it;
351 it = it->GetNext() )
352 {
353 it->GetData()->Wait();
354 delete it->GetData();
355 }
356
357 for ( EList::compatibility_iterator it2 = m_eventWorkers.GetFirst();
358 it2;
359 it2->GetNext() )
360 {
361 delete it2->GetData();
362 }
363
364 m_threadWorkers.Clear();
365 m_eventWorkers.Clear();
366 m_listeningSocket->Destroy();
367 return 0;
368 }
369
370 void Server::OnSocketEvent(wxSocketEvent& pEvent)
371 {
372 switch(pEvent.GetSocketEvent())
373 {
374 case wxSOCKET_INPUT:
375 wxLogError("Unexpected wxSOCKET_INPUT in wxSocketServer");
376 break;
377 case wxSOCKET_OUTPUT:
378 wxLogError("Unexpected wxSOCKET_OUTPUT in wxSocketServer");
379 break;
380 case wxSOCKET_CONNECTION:
381 {
382 wxSocketBase* sock = m_listeningSocket->Accept();
383 wxIPV4address addr;
384 if (!sock->GetPeer(addr))
385 {
386 wxLogError("Server: cannot get peer info");
387 } else {
388 wxLogMessage("Got connection from %s:%d",addr.IPAddress().c_str(), addr.Service());
389 }
390 bool createThread;
391
392 if (m_workMode != MIXED)
393 createThread = m_workMode == THREADS;
394 else
395 createThread = (wxDateTime::Now().GetSecond())%2 == 0;
396
397 if (createThread)
398 {
399 ThreadWorker* c = new ThreadWorker(sock);
400 if (c->Create() == wxTHREAD_NO_ERROR)
401 {
402 m_threadWorkers.Append(c);
403 if (m_threadWorkers.GetCount() > m_maxThreadWorkers)
404 m_maxThreadWorkers++;
405 m_threadWorkersCreated++;
406 c->Run();
407 }
408 else
409 {
410 wxLogError("Server: cannot create next thread (current threads: %d", m_threadWorkers.size());
411 };
412 }
413 else
414 {
415 EventWorker* w = new EventWorker(sock);
416 m_eventWorkers.Append(w);
417 if (m_eventWorkers.GetCount() > m_maxEventWorkers)
418 m_maxEventWorkers++;
419 m_eventWorkersCreated++;
420 }
421 }
422 break;
423 case wxSOCKET_LOST:
424 wxLogError("Unexpected wxSOCKET_LOST in wxSocketServer");
425 break;
426 }
427 }
428
429 void Server::OnWorkerEvent(WorkerEvent& pEvent)
430 {
431 //wxLogMessage("Got worker event");
432 for(TList::compatibility_iterator it = m_threadWorkers.GetFirst(); it ; it = it->GetNext())
433 {
434 if (it->GetData() == pEvent.m_sender)
435 {
436 wxLogVerbose("Deleting thread worker (%lu left)",
437 static_cast<unsigned long>( m_threadWorkers.GetCount() ));
438 it->GetData()->Wait();
439 delete it->GetData();
440 m_threadWorkers.DeleteNode(it);
441 if (!pEvent.m_workerFailed)
442 m_threadWorkersDone++;
443 else
444 m_threadWorkersFailed++;
445 break;
446 }
447 }
448 for(EList::compatibility_iterator it2 = m_eventWorkers.GetFirst(); it2 ; it2 = it2->GetNext())
449 {
450 if (it2->GetData() == pEvent.m_sender)
451 {
452 wxLogVerbose("Deleting event worker (%lu left)",
453 static_cast<unsigned long>( m_eventWorkers.GetCount() ));
454 delete it2->GetData();
455 m_eventWorkers.DeleteNode(it2);
456 if (!pEvent.m_workerFailed)
457 m_eventWorkersDone++;
458 else
459 m_eventWorkersFailed++;
460 break;
461 }
462 }
463
464 if (m_eventWorkers.GetCount() == 0 && m_threadWorkers.GetCount() == 0)
465 {
466 mTimer.Start(1000,true);
467 }
468 }
469
470 void Server::OnTimerEvent(wxTimerEvent&)
471 {
472 DumpStatistics();
473 }
474
475
476 BEGIN_EVENT_TABLE(Server,wxEvtHandler)
477 EVT_SOCKET(wxID_ANY,Server::OnSocketEvent)
478 EVT_WORKER(Server::OnWorkerEvent)
479 EVT_TIMER(wxID_ANY,Server::OnTimerEvent)
480 END_EVENT_TABLE()
481
482
483 ThreadWorker::ThreadWorker(wxSocketBase* pSocket) : wxThread(wxTHREAD_JOINABLE)
484 {
485 m_socket = pSocket;
486 //Notify() cannot be called in thread context. We have to detach from main loop
487 //before switching thread contexts.
488 m_socket->Notify(false);
489 m_socket->SetFlags(wxSOCKET_WAITALL|wxSOCKET_BLOCK);
490 pSocket->GetPeer(m_peer);
491 }
492
493 wxThread::ExitCode ThreadWorker::Entry()
494 {
495 WorkerEvent e(this);
496 if (!m_socket->IsConnected())
497 {
498 LogWorker("ThreadWorker: not connected",wxLOG_Error);
499 return 0;
500 }
501 int to_process = -1;
502 if (m_socket->IsConnected())
503 {
504 unsigned char signature[2];
505 LogWorker("ThreadWorker: reading for data");
506 to_process = 2;
507 do
508 {
509 m_socket->Read(&signature,to_process);
510 if (m_socket->Error())
511 {
512 LogWorker("ThreadWorker: Read error",wxLOG_Error);
513 wxGetApp().AddPendingEvent(e);
514 return 0;
515 }
516 to_process -= m_socket->LastCount();
517 LogWorker(wxString::Format("to_process: %d",to_process));
518
519 }
520 while (!m_socket->Error() && to_process != 0);
521
522 if (signature[0] == 0)
523 {
524 e.m_exit = true;
525 return 0;
526 }
527
528 if (signature[0] == 0xCE)
529 {
530 LogWorker("This server does not support test2 from GUI client",wxLOG_Error);
531 e.m_workerFailed = true;
532 e.m_exit = true;
533 return 0;
534 }
535 int size = signature[1] * (signature[0] == 0xBE ? 1 : 1024);
536 char* buf = new char[size];
537 LogWorker(wxString::Format("Message signature: chunks: %d, kilobytes: %d, size: %d (bytes)",signature[0],signature[1],size));
538
539 to_process = size;
540 LogWorker(wxString::Format("ThreadWorker: reading %d bytes of data",to_process));
541
542 do
543 {
544 m_socket->Read(buf,to_process);
545 if (m_socket->Error())
546 {
547 LogWorker("ThreadWorker: Read error",wxLOG_Error);
548 wxGetApp().AddPendingEvent(e);
549 return 0;
550 }
551 to_process -= m_socket->LastCount();
552 LogWorker(wxString::Format("ThreadWorker: %d bytes readed, %d todo",m_socket->LastCount(),to_process));
553
554 }
555 while(!m_socket->Error() && to_process != 0);
556
557 to_process = size;
558
559 do
560 {
561 m_socket->Write(buf,to_process);
562 if (m_socket->Error())
563 {
564 LogWorker("ThreadWorker: Write error",wxLOG_Error);
565 break;
566 }
567 to_process -= m_socket->LastCount();
568 LogWorker(wxString::Format("ThreadWorker: %d bytes written, %d todo",m_socket->LastCount(),to_process));
569 }
570 while(!m_socket->Error() && to_process != 0);
571 }
572
573 LogWorker("ThreadWorker: done");
574 e.m_workerFailed = to_process != 0;
575 m_socket->Destroy();
576 wxGetApp().AddPendingEvent(e);
577 return 0;
578 }
579
580 EventWorker::EventWorker(wxSocketBase* pSock)
581 : m_socket(pSock),
582 m_inbuf(NULL),
583 m_infill(0),
584 m_outbuf(NULL),
585 m_outfill(0)
586 {
587 m_socket->SetNotify(wxSOCKET_LOST_FLAG|wxSOCKET_INPUT_FLAG|wxSOCKET_OUTPUT_FLAG);
588 m_socket->Notify(true);
589 m_socket->SetEventHandler(*this);
590 m_socket->SetFlags(wxSOCKET_NOWAIT);
591 m_socket->GetPeer(m_peer);
592 }
593
594 EventWorker::~EventWorker()
595 {
596 m_socket->Destroy();
597 delete [] m_inbuf;
598 delete [] m_outbuf;
599 }
600
601 void
602 EventWorker::DoRead()
603 {
604 if (m_inbuf == NULL)
605 {
606 //read message header
607 do
608 {
609 m_socket->Read(m_signature + m_infill, 2 - m_infill);
610 if (m_socket->Error())
611 {
612 if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
613 {
614 LogWorker(wxString::Format("Read error (%d): %s",m_socket->LastError(),GetSocketErrorMsg(m_socket->LastError())),wxLOG_Error);
615 m_socket->Close();
616 }
617 }
618 else
619 {
620 m_infill += m_socket->LastCount();
621 if (m_infill == 2)
622 {
623 unsigned char chunks = m_signature[1];
624 unsigned char type = m_signature[0];
625 if (type == 0xCE)
626 {
627 LogWorker("This server does not support test2 from GUI client",wxLOG_Error);
628 m_written = -1; //wxSOCKET_LOST will interpret this as failure
629 m_socket->Close();
630 }
631 else if (type == 0xBE || type == 0xDE)
632 {
633 m_size = chunks * (type == 0xBE ? 1 : 1024);
634 m_inbuf = new char[m_size];
635 m_outbuf = new char[m_size];
636 m_infill = 0;
637 m_outfill = 0;
638 m_written = 0;
639 LogWorker(wxString::Format("Message signature: len: %d, type: %s, size: %d (bytes)",chunks,type == 0xBE ? "b" : "kB",m_size));
640 break;
641 }
642 else
643 {
644 LogWorker(wxString::Format("Unknown test type %x",type));
645 m_socket->Close();
646 }
647 }
648 }
649 }
650 while(!m_socket->Error() && (2 - m_infill != 0));
651 }
652
653 if (m_inbuf == NULL)
654 return;
655 //read message data
656 do
657 {
658 if (m_size == m_infill)
659 {
660 m_signature[0] = m_signature[1] = 0x0;
661 wxDELETEA(m_inbuf);
662 m_infill = 0;
663 return;
664 }
665 m_socket->Read(m_inbuf + m_infill,m_size - m_infill);
666 if (m_socket->Error())
667 {
668 if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
669 {
670 LogWorker(wxString::Format("Read error (%d): %s",
671 m_socket->LastError(),
672 GetSocketErrorMsg(m_socket->LastError())),
673 wxLOG_Error);
674
675 m_socket->Close();
676 }
677 }
678 else
679 {
680 memcpy(m_outbuf+m_outfill,m_inbuf+m_infill,m_socket->LastCount());
681 m_infill += m_socket->LastCount();
682 m_outfill += m_socket->LastCount();
683 DoWrite();
684 }
685 }
686 while(!m_socket->Error());
687 };
688
689 void EventWorker::OnSocketEvent(wxSocketEvent& pEvent)
690 {
691 switch(pEvent.GetSocketEvent())
692 {
693 case wxSOCKET_INPUT:
694 DoRead();
695 break;
696
697 case wxSOCKET_OUTPUT:
698 if ( m_outbuf )
699 DoWrite();
700 break;
701
702 case wxSOCKET_CONNECTION:
703 LogWorker("Unexpected wxSOCKET_CONNECTION in EventWorker", wxLOG_Error);
704 break;
705
706 case wxSOCKET_LOST:
707 {
708 LogWorker("Connection lost");
709 WorkerEvent e(this);
710 e.m_workerFailed = m_written != m_size;
711 wxGetApp().AddPendingEvent(e);
712 }
713 break;
714 }
715 }
716
717 void EventWorker::DoWrite()
718 {
719 do
720 {
721 if (m_written == m_size)
722 {
723 wxDELETEA(m_outbuf);
724 m_outfill = 0;
725 LogWorker( "All data written");
726 return;
727 }
728 if (m_outfill - m_written == 0)
729 {
730 return;
731 }
732 m_socket->Write(m_outbuf + m_written,m_outfill - m_written);
733 if (m_socket->Error())
734 {
735 if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
736 {
737 LogWorker(
738 wxString::Format("Write error (%d): %s",
739 m_socket->LastError(),
740 GetSocketErrorMsg(m_socket->LastError())
741 )
742 ,wxLOG_Error
743 );
744 m_socket->Close();
745 }
746 else
747 {
748 LogWorker("Write would block, waiting for OUTPUT event");
749 }
750 }
751 else
752 {
753 memmove(m_outbuf,m_outbuf+m_socket->LastCount(),m_outfill-m_socket->LastCount());
754 m_written += m_socket->LastCount();
755 }
756 LogWorker(wxString::Format("Written %d of %d bytes, todo %d",
757 m_socket->LastCount(),m_size,m_size - m_written));
758 }
759 while (!m_socket->Error());
760 }
761
762 BEGIN_EVENT_TABLE(EventWorker,wxEvtHandler)
763 EVT_SOCKET(wxID_ANY,EventWorker::OnSocketEvent)
764 END_EVENT_TABLE()
wxWidgets (Impactor)