Merge branch 'master' of feistymeow.org:feisty_meow
[feisty_meow.git] / octopi / library / sockets / socket_minder.cpp
1 /*****************************************************************************\
2 *                                                                             *
3 *  Name   : socket_minder                                                     *
4 *  Author : Chris Koeritz                                                     *
5 *                                                                             *
6 *******************************************************************************
7 * Copyright (c) 1999-$now By Author.  This program is free software; you can  *
8 * redistribute it and/or modify it under the terms of the GNU General Public  *
9 * License as published by the Free Software Foundation; either version 2 of   *
10 * the License or (at your option) any later version.  This is online at:      *
11 *     http://www.fsf.org/copyleft/gpl.html                                    *
12 * Please send any updates to: fred@gruntose.com                               *
13 \*****************************************************************************/
14
15 #include "raw_socket.h"
16 #include "socket_data.h"
17 #include "socket_minder.h"
18 #include "tcpip_stack.h"
19
20 #include <basis/mutex.h>
21 #include <loggers/critical_events.h>
22 #include <loggers/program_wide_logger.h>
23 #include <processes/ethread.h>
24 #include <processes/os_event.h>
25 #include <structures/set.h>
26 #include <structures/amorph.h>
27 #include <structures/unique_id.h>
28 #include <textual/parser_bits.h>
29
30 #include <errno.h>
31 #ifdef __WIN32__
32   #include <ws2tcpip.h>
33 #endif
34 #ifdef __UNIX__
35   #include <arpa/inet.h>
36   #include <sys/socket.h>
37 #endif
38
39 using namespace basis;
40 using namespace loggers;
41 using namespace processes;
42 using namespace structures;
43 using namespace textual;
44 using namespace timely;
45
46 namespace sockets {
47
48 //#define DEBUG_SOCKET_MINDER
49   // uncomment for noisiness.
50
51 #undef LOG
52 #define LOG(to_print) CLASS_EMERGENCY_LOG(program_wide_logger::get(), to_print)
53
54 const int SOCKET_CHECK_INTERVAL = 50;
55   // we will scoot around in our sockets this frequently.
56
57 const int SOCKMIND_MAXIMUM_RECEIVES = 10;
58   // we'll receive this many items from the socket in one go.
59
60 const int MAXIMUM_TRANSFER_CHUNK = 512 * KILOBYTE;
61   // largest block of data we'll try to deal with at one time.
62
63 const int CONN_ALERT_INTERVAL = 100;
64   // this is the most frequently that we will generate a connection checking
65   // event.
66
67 const int MULTI_SELECT_TIMEOUT = 250;
68   // the snooze on select will occur for this long.  during this interval,
69   // it is likely no new sockets will be considered.
70
71 //////////////
72
73 class socket_data_amorph : public amorph<socket_data> {};
74
75 //////////////
76
77 class socket_minder_prompter : public ethread
78 {
79 public:
80   socket_minder_prompter(socket_minder &parent)
81   : ethread(SOCKET_CHECK_INTERVAL, ethread::SLACK_INTERVAL),
82     _parent(parent)
83   {
84     start(NULL_POINTER);
85   }
86
87   ~socket_minder_prompter() {
88     stop();  // shut down our thread.
89   }
90
91   virtual void perform_activity(void *formal(ptr)) { _parent.snoozy_select(); }
92
93 private:
94   socket_minder &_parent;  // the object we're hooked to.
95 };
96
97 //////////////
98
99 socket_minder::socket_minder(post_office &post, int parent_route,
100     int event_type, int message)
101 : _post(post),
102   _parent_route(parent_route),
103   _event_type(event_type),
104   _lock(new mutex),
105   _socket_list(new socket_data_amorph),
106   _socks(new raw_socket),
107   _stack(new tcpip_stack),
108   _message(message),
109   _pending_sox(new int_set),
110   _prompter(new socket_minder_prompter(*this))
111 {
112   _prompter->start(NULL_POINTER);
113 }
114
115 socket_minder::~socket_minder()
116 {
117   _prompter->stop();
118   WHACK(_prompter);
119   WHACK(_socket_list);
120   WHACK(_lock);
121   WHACK(_pending_sox);
122   WHACK(_socks); 
123   WHACK(_stack);
124 }
125
126 void socket_minder::disengage()
127 {
128   _prompter->stop();
129 }
130
131 astring socket_minder::text_form() const
132 {
133   auto_synchronizer l(*_lock);
134   astring to_return;
135
136   for (int i = 0; i < _socket_list->elements(); i++) {
137     const socket_data *curr = _socket_list->get(i);
138     to_return += curr->text_form();
139     if (i != _socket_list->elements() - 1)
140       to_return += parser_bits::platform_eol_to_chars();
141   }
142
143   return to_return;
144 }
145
146 void socket_minder::snoozy_select()
147 {
148   FUNCDEF("snoozy_select");
149   int_array read_sox;
150   int_array write_sox;
151   int_array pending;
152
153   get_sockets(read_sox, write_sox, pending);
154
155   // process any with pending connections right now, rather than later.
156   for (int p = 0; p < pending.length(); p++) {
157     socket_data *sd = lock_socket_data(pending[p]);
158     if (!sd) continue;  // something hosed there.
159     handle_pending_connecters(*sd);
160     unlock_socket_data(sd);
161   }
162
163   // now select on all of our sockets simultaneously.
164   int ret = _socks->select(read_sox, write_sox, MULTI_SELECT_TIMEOUT);
165   if (!ret || (!read_sox.length() && !write_sox.length()) ) {
166     return;  // nothing happened.
167   }
168
169   // rotate through the lists and push socket_minders around as needed.
170   // any sockets we have events for but no socket_data are orphans and will
171   // be ignored.
172
173   // check read sockets.
174   for (int r = 0; r < read_sox.length(); r++) {
175     const int sock = read_sox[r];
176     if (owns_socket(sock)) {
177       socket_data *sd = lock_socket_data(sock);
178       if (!sd) continue;  // something hosed there.
179       push_receives(*sd, SI_READABLE);
180       unlock_socket_data(sd);
181       read_sox.zap(r, r);
182       r--;  // skip back before deleted guy.
183     }
184   }
185
186   // check write sockets.
187   for (int w = 0; w < write_sox.length(); w++) {
188     const int sock = write_sox[w];
189     if (owns_socket(sock)) {
190       socket_data *sd = lock_socket_data(sock);
191       if (!sd) continue;  // something hosed there.
192       push_sends(*sd, SI_WRITABLE);
193       unlock_socket_data(sd);
194       write_sox.zap(w, w);
195       w--;  // skip back before deleted guy.
196     }
197   }
198 }
199
200 void socket_minder::get_sockets(int_array &read_sox, int_array &write_sox,
201     int_array &pendings) const
202 {
203   auto_synchronizer l(*_lock);
204   for (int i = 0; i < _socket_list->elements(); i++) {
205     socket_data *sd = _socket_list->borrow(i);
206     if (sd->_connection_pending) {
207       // this is not ready for sends and receives yet.
208       pendings += sd->_socket;
209     } else {
210       // always add sockets to test if they have data waiting.
211       read_sox += sd->_socket;
212       // only check on writability if there is data pending for sending.
213       if (sd->_partially_sent.length())
214         write_sox += sd->_socket;
215     }
216   }
217 }
218
219 bool socket_minder::owns_socket(int socket) const
220 {
221   auto_synchronizer l(*_lock);
222   for (int i = 0; i < _socket_list->elements(); i++) {
223     if (_socket_list->borrow(i)->_socket == socket) return true;
224   }
225   return false;
226 }
227
228 socket_data *socket_minder::lock_socket_data(int socket)
229 {
230   _lock->lock();
231   for (int i = 0; i < _socket_list->elements(); i++)
232     if (_socket_list->borrow(i)->_socket == socket)
233       return _socket_list->borrow(i);
234   // this is a failure to get here; there was no suitable socket.
235   _lock->unlock();
236   return NULL_POINTER;
237 }
238
239 void socket_minder::unlock_socket_data(socket_data *to_unlock)
240 {
241   if (!to_unlock) return;
242 //can't affect it now.  to_unlock = NULL_POINTER;
243   _lock->unlock();
244 }
245
246 bool socket_minder::add_socket_data(int socket, bool server, int server_socket,
247     bool connected_mode, bool connection_pending)
248 {
249   auto_synchronizer l(*_lock);
250   socket_data *harpo = lock_socket_data(socket);
251   if (harpo) {
252     unlock_socket_data(harpo);
253     return false;
254   }
255   socket_data *new_data = new socket_data(socket, server, server_socket,
256       connected_mode);
257   _socks->set_non_blocking(socket);
258     // ensure the new guy is treated as non-blocking.  unix does not seem
259     // to inherit this from the parent.
260   new_data->_connection_pending = connection_pending;
261   _socket_list->append(new_data);
262   return true;
263 }
264
265 bool socket_minder::remove_socket_data(int socket)
266 {
267   FUNCDEF("remove_socket_data");
268   auto_synchronizer l(*_lock);
269   for (int i = 0; i < _socket_list->elements(); i++) {
270     if (_socket_list->borrow(i)->_socket == socket) {
271       // give the socket a last chance to get its data out.
272       evaluate_interest(*_socket_list->borrow(i));
273       _socket_list->zap(i, i);
274       return true;
275     }
276   }
277 //  LOG(a_sprintf("couldn't find socket %d.", socket));
278   return false;
279 }
280
281 bool socket_minder::register_interest(int socket, int interests)
282 {
283 #ifdef DEBUG_SOCKET_MINDER
284   FUNCDEF("register_interest");
285 #endif
286   socket_data *harpo = lock_socket_data(socket);
287 #ifdef DEBUG_SOCKET_MINDER
288   LOG(astring(astring::SPRINTF, "registering interest of %d for socket "
289       "%d.", interests, socket));
290 #endif
291   if (!harpo) return false;
292   harpo->_registered_interests = interests;
293   unlock_socket_data(harpo);
294   return true;
295 }
296
297 bool socket_minder::is_connection_pending(int socket)
298 {
299   socket_data *harpo = lock_socket_data(socket);
300   if (!harpo) return false;
301   bool to_return = harpo->_connection_pending;
302   unlock_socket_data(harpo);
303   return to_return;
304 }
305
306 bool socket_minder::set_connection_pending(int socket, bool pending)
307 {
308   socket_data *harpo = lock_socket_data(socket);
309   if (!harpo) return false;
310   harpo->_connection_pending = pending;
311   unlock_socket_data(harpo);
312   return true;
313 }
314
315 void socket_minder::fire_event(int to_fire, int at_whom,
316     basis::un_int parm1, basis::un_int parm2)
317 {
318   _post.drop_off(at_whom, new OS_event(_event_type, to_fire, parm1, parm2));
319 }
320
321 void socket_minder::put_pending_server(int to_put, bool at_head)
322 {
323   if (!to_put) return;  // bogus.
324   auto_synchronizer l(*_lock);
325   if (at_head) {
326     _pending_sox->insert(0, 1);
327     (*_pending_sox)[0] = to_put;
328   } else {
329     *_pending_sox += to_put;
330   }
331 }
332
333 bool socket_minder::zap_pending_server(int socket)
334 {
335   auto_synchronizer l(*_lock);
336   if (!_pending_sox->member(socket)) return false;
337   _pending_sox->remove(socket);
338   return true;
339 }
340
341 int socket_minder::get_pending_server()
342 {
343   auto_synchronizer l(*_lock);
344   if (!_pending_sox->length()) return 0;
345   int to_return = _pending_sox->get(0);
346   _pending_sox->zap(0, 0);
347   return to_return;
348 }
349
350 bool socket_minder::handle_pending_connecters(socket_data &to_peek)
351 {
352   FUNCDEF("handle_pending_connecters");
353   if (!to_peek._connection_pending) return false;  // not needed here.
354
355   if (to_peek._last_conn_alert > time_stamp(-CONN_ALERT_INTERVAL)) {
356     // not time yet.
357     return false;
358   }
359   to_peek._last_conn_alert.reset();
360
361   // force the issue; there is no simple way to portably know whether
362   // the socket got a connection or not, so just say it did.
363   if (!to_peek._is_server
364       && (to_peek._registered_interests & SI_CONNECTED) ) {
365     // deal with a client first.  this just means, zing an event.
366 #ifdef DEBUG_SOCKET_MINDER
367     LOG(a_sprintf("sending client SI_CONNECTED event on parent %d",
368         _parent_route));
369 #endif
370     fire_event(_message, _parent_route, to_peek._socket, SI_CONNECTED);
371   } else if (to_peek._is_server
372       && (to_peek._registered_interests & SI_CONNECTED) ) {
373     // special handling for servers.  we accept the waiting guy if he's
374     // there.  otherwise we don't send the event.
375     sockaddr sock_addr;
376     socklen_t sock_len = sizeof(sock_addr);
377     int new_sock = int(::accept(to_peek._socket, &sock_addr, &sock_len));
378       // check for a new socket.
379     if (new_sock != INVALID_SOCKET) {
380       LOG(a_sprintf("accept got a client socket %d.", new_sock));
381       if (_pending_sox->member(new_sock)) {
382         LOG(a_sprintf("already have seen socket %d in pending!", new_sock));
383       } else {
384         *_pending_sox += new_sock;
385 #ifdef DEBUG_SOCKET_MINDER
386         LOG(a_sprintf("sending server SI_CONNECTED event on parent %d",
387             _parent_route));
388 #endif
389         fire_event(_message, _parent_route, to_peek._socket, SI_CONNECTED);
390       }
391     } else if (_pending_sox->length()) {
392       // there are still pending connectees.
393       fire_event(_message, _parent_route, to_peek._socket, SI_CONNECTED);
394     }
395   }
396   // also, if the connection is still pending, we don't want to select on
397   // it yet.
398   return true;
399 }
400
401 bool socket_minder::evaluate_interest(socket_data &to_peek)
402 {
403   FUNCDEF("evaluate_interest");
404   if (to_peek._connection_pending) {
405     return handle_pending_connecters(to_peek);
406   }
407
408   int sel_mode = 0;
409
410   int states = _socks->select(to_peek._socket, sel_mode);
411
412   if (!states) {
413     return true;  // nothing to report.
414   }
415
416   if (! (states & SI_ERRONEOUS) && ! (states & SI_DISCONNECTED) ) {
417     push_sends(to_peek, states);
418     push_receives(to_peek, states);
419   }
420
421   if ( (to_peek._registered_interests & SI_ERRONEOUS)
422       && (states & SI_ERRONEOUS) ) {
423     // there's some kind of bad problem on the socket.
424     LOG(a_sprintf("socket %d has status of erroneous.", to_peek._socket));
425 //hmmm: what to do?  generate an event?
426   }
427
428   if ( (to_peek._registered_interests & SI_DISCONNECTED)
429       && (states & SI_DISCONNECTED) ) {
430     // we lost our state of connectedness.
431     fire_event(_message, _parent_route, to_peek._socket,
432         SI_DISCONNECTED);
433     return true;  // get out now.
434   }
435
436   return true;
437 }
438
439 void socket_minder::push_sends(socket_data &to_poke, int states)
440 {
441   FUNCDEF("push_sends");
442   if (to_poke._connection_pending) {
443     LOG("not supposed to try this when not connected yet...");
444   }
445
446 #ifdef DEBUG_SOCKET_MINDER
447   if (to_poke._partially_sent.length()) {
448     LOG(a_sprintf("socket %d: %d bytes to send.", to_poke._socket,
449         to_poke._partially_sent.length()));
450   }
451 #endif
452
453   int error_code = 0;
454
455   if ( (states & SI_WRITABLE) && to_poke._partially_sent.length()) {
456     // write to the socket since we see an opportunity.
457     byte_array &to_send = to_poke._partially_sent;
458     int len_sent = send(to_poke._socket, (char *)to_send.observe(),
459         to_send.length(), 0);
460     if (!len_sent) {
461       // nothing got sent.
462       LOG(a_sprintf("didn't send any data on socket %d.", to_poke._socket));
463     } else if (len_sent == SOCKET_ERROR) {
464       // something bad happened.
465       error_code = critical_events::system_error();
466     } else {
467 #ifdef DEBUG_SOCKET_MINDER
468       LOG(a_sprintf("updating that %d bytes got sent out of %d to send.",
469           len_sent, to_send.length()));
470       if (to_send.length() != len_sent)
471         LOG(a_sprintf("size to send (%d) not same as size sent (%d).",
472             to_send.length(), len_sent));
473 #endif
474       // update the partially sent chunk for the bit we sent out.
475       to_send.zap(0, len_sent - 1);
476     }
477   }
478
479   // handle errors we have seen.
480   if (error_code) {
481     if (error_code != SOCK_EWOULDBLOCK)
482       LOG(astring(astring::SPRINTF, "error on socket %d is %s.",
483           to_poke._socket, _stack->tcpip_error_name(error_code).s()));
484
485     switch (error_code) {
486       case SOCK_ENOTSOCK:  // fall-through.
487       case SOCK_ECONNABORTED:  // fall-through.
488       case SOCK_ECONNRESET: {
489         // the connection got hammerlocked somehow.
490         LOG(a_sprintf("due to %s condition, closing socket %d.",
491             _stack->tcpip_error_name(error_code).s(), to_poke._socket));
492         fire_event(_message, _parent_route, to_poke._socket,
493             SI_DISCONNECTED);
494         to_poke._partially_sent.reset();  // clear with no connection.
495         break;
496       }
497     }
498   }
499 }
500
501 void socket_minder::push_receives(socket_data &to_poke, int states)
502 {
503   FUNCDEF("push_receives");
504   if (to_poke._connection_pending) {
505     LOG("not supposed to try this when not connected yet...");
506   }
507
508 #ifdef DEBUG_SOCKET_MINDER
509   if (to_poke._partially_received.length())
510     LOG(a_sprintf("socket %d: %d bytes already received.", to_poke._socket,
511         to_poke._partially_received.length()));
512 #endif
513
514   int error_code = 0;
515
516   if ( (states & SI_READABLE) && to_poke._connected_mode) {
517     // grab any data that's waiting on the connection-oriented socket.
518
519     bool got_something = true;  // hopeful for now.
520     to_poke._receive_buffer.reset(MAXIMUM_TRANSFER_CHUNK);
521       // allocate space where we'll get new data.
522     int count = 0;
523     while (got_something && (count++ < SOCKMIND_MAXIMUM_RECEIVES)) {
524       got_something = false;  // now get all pessimistic.
525       int len = recv(to_poke._socket, (char *)to_poke._receive_buffer.observe(),
526           to_poke._receive_buffer.length(), 0);
527       if (len > 0) {
528 #ifdef DEBUG_SOCKET_MINDER
529         LOG(a_sprintf("received %d bytes on socket %d.", len, to_poke._socket));
530 #endif
531         // we received some actual data; we're happy again.
532         got_something = true;
533         // zap any extra off.
534         if (len < MAXIMUM_TRANSFER_CHUNK)
535           to_poke._receive_buffer.zap(len, to_poke._receive_buffer.last());
536         // add in what we were given.
537         to_poke._partially_received += to_poke._receive_buffer;
538         to_poke._receive_buffer.reset(MAXIMUM_TRANSFER_CHUNK);
539           // reset to the right size for trying some more.
540       } else {
541         error_code = critical_events::system_error();
542
543         // reset the states flag based on current state.
544         states = _socks->select(to_poke._socket,
545             raw_socket::SELECTING_JUST_READ);
546         if (states & SI_DISCONNECTED) {
547           error_code = SOCK_ECONNRESET;  // make like regular disconnect.
548           LOG(a_sprintf("noticed disconnection on socket %d.",
549               to_poke._socket));
550           // close the socket; we use a temporary because the close method
551           // wants to actually store zero in the socket.
552           basis::un_int deader = to_poke._socket;
553           _socks->close(deader);
554         }
555
556       }
557     }
558
559     if ( !(states & SI_DISCONNECTED)
560         && to_poke._partially_received.length()) { 
561 #ifdef DEBUG_SOCKET_MINDER
562       LOG(a_sprintf("firing readable now: sock=%d", to_poke._socket));
563 #endif
564       fire_event(_message, _parent_route, to_poke._socket, SI_READABLE);
565     }
566   } else if ( (states & SI_READABLE) && !to_poke._connected_mode) {
567     // datagram style; we need to still alert the parent.
568     fire_event(_message, _parent_route, to_poke._socket,
569         SI_READABLE);
570   }
571
572   // handle errors we have seen.
573   if (error_code) {
574     if (error_code != SOCK_EWOULDBLOCK)
575       LOG(astring(astring::SPRINTF, "error on socket %d is %s.",
576           to_poke._socket, _stack->tcpip_error_name(error_code).s()));
577
578     switch (error_code) {
579       case SOCK_ENOTSOCK:  // fall-through.
580       case SOCK_ECONNABORTED:  // fall-through.
581       case SOCK_ECONNRESET: {
582         // the connection got hammerlocked somehow.
583         LOG(a_sprintf("due to %s condition, closing socket %d.",
584             _stack->tcpip_error_name(error_code).s(), to_poke._socket));
585         fire_event(_message, _parent_route, to_poke._socket,
586             SI_DISCONNECTED);
587         basis::un_int deader = to_poke._socket;
588         _socks->close(deader);
589         to_poke._partially_sent.reset();  // clear with no connection.
590         break;
591       }
592     }
593   }
594 }
595
596 } //namespace.
597