more patches for spread 5.0.0rc1
"Daniel F. Savarese" <[email protected]> Fri, 06 Jan 2017 21:38:21 -0500
| Newsgroups | gmane.network.spread.user |
|---|---|
| Message-ID | <[email protected]> |
In message <[email protected]>, John Lane Schultz writes: >Thank you for testing and your detailed patch notes! I'll incorporate them int >o the next release candidate, which should be out in the next week or two. Thanks! Just make sure you sanity check them based on your superior knowledge of the code. I spoke too soon when I said libssrcspread passed all unit tests with vanilla spread-5.0.0rc1 (no patches). One of the test suites started failing sporadically. A mailbox in one test would join a group and send a message to the group, but leave the group and disconnect before receiving the message. In the immediately subsequent test, a new mailbox performing a receive would receive the message for the old mailbox even though the new mailbox had a different private group name and did not belong to the group that the original message was sent to. I traced the source of the problem to two close() calls on the mailbox socket descriptors in session.c. The calls in Sess_read() and Sess_kill() close a session's socket descriptor before the session is removed. That frees the descriptor to be assigned on a future accept call. If a new session is accepted before the old one is removed, recycling the old descriptor, then the new session ends up receiving the old session's group messages. Any calls to get_session_index will return the new session. Furthermore, the new session can be the recipient of any outstanding queued Sess_badger_TO calls if more than one was queued for the old session. The solution is to close the session descriptor when the session is removed by Sess_remove_session() and to dequeue all events associated with the descriptor, not just one Sess_badger_TO. The bug epends on timing (an accept has to happen after a disconnect but before a self-leave or other group message is delivered) and therefore likely rarely encountered. I only spotted it because of short unit tests making very fast connections/disconnections. As part of tracking down the problem, I made some minor refactorings that don't change any behavior. Therefore my patch isn't as succinct as it could otherwise be. If my changes look correct, cherry-pick the essential portions if the cosmetic changes aren't appropriate. I tried to maintain the same coding style as the original code, but Sess_clear_session() differs. Patches follow with explanatory commit messages. Patch 1: ------- Removed no-op protocol functions. Prot_kill_session, Prot_Create_Local, and Prot_Destroy are all empty functions that do nothing and simply return. diff --git a/daemon/protocol.c b/daemon/protocol.c index 257d833..08afe48 100644 --- a/daemon/protocol.c +++ b/daemon/protocol.c @@ -233,21 +233,6 @@ void Prot_set_down_queue( int queue_type ) } } -void Prot_Create_Local_Session( session *new_sess ) -{ - return; -} - -void Prot_Destroy_Local_Session( session *old_sess ) -{ - return; -} - -void Prot_kill_session( message_obj *msg ) -{ - return; -} - down_link *Prot_Create_Down_Link( message_obj *msg, int type, int mbox, int cur_element ) { down_link *down_ptr; diff --git a/daemon/protocol.h b/daemon/protocol.h index 36123f4..6c104f4 100644 --- a/daemon/protocol.h +++ b/daemon/protocol.h @@ -61,10 +61,7 @@ void Prot_init(void); void Prot_set_down_queue( int queue_type ); void Prot_new_message( down_link *down_ptr, int not_used_in_spread3_p ); void Prot_init_down_queues(void); -void Prot_Create_Local_Session(session *new_sess); -void Prot_Destroy_Local_Session(session *old_sess); down_link *Prot_Create_Down_Link(message_obj *msg, int type, int mbox, int cur_element); -void Prot_kill_session(message_obj *msg); void Prot_set_prev_proc(configuration *memb); /* thresholds defined in net_types.h: UNRELIABLE_TYPE, AGREED_TYPE, BLOCK_REGULAR_DELIVERY, etc. */ diff --git a/daemon/session.c b/daemon/session.c index 8121f96..03af612 100644 --- a/daemon/session.c +++ b/daemon/session.c @@ -1044,8 +1044,6 @@ void Sess_session_authorized(int ses) Sessions[ses].status = Set_op_session( Sessions[ses].status ); Sessions[ses].status = Clear_preauth_session( Sessions[ses].status ); - Prot_Create_Local_Session(&Sessions[ses]); - Message_reset_current_location(&(Sessions[ses].read) ); Message_reset_current_location(&(Sessions[ses].write) ); Sessions[ses].read.in_mess_head = 1; @@ -1967,13 +1965,11 @@ static void Sess_handle_kill( message_link *mess_link ) if( Is_op_session( Sessions[ses].status ) ) Alarm( EXIT, "Sess_handle_kill: killing unkilled session bug!\n"); Sess_unhash_session (&Sessions[ses]); - Prot_Destroy_Local_Session(&(Sessions[ses]) ); Sess_remove_session (&Sessions[ses]); Num_sessions--; GlobalStatus.num_sessions = Num_sessions; } } - Prot_kill_session(mess_link->mess); Sess_dispose_message( mess_link ); } commit d7fdb230807213456161e7e5bc75c8d3b46aee23 Author: Daniel F. Savarese <[email protected]> Date: Wed Jan 4 13:18:40 2017 -0500 Moved Sess_unash_session() calls into Sess_remove_session(). Sess_unhash_session() is always called immediately before Sess_remove_session(). I simplified the call points by moving the calls to the first line of Sess_remove_session(). diff --git a/daemon/session.c b/daemon/session.c index 03af612..b7b5ba9 100644 --- a/daemon/session.c +++ b/daemon/session.c @@ -228,6 +228,8 @@ static int Sess_insert_new_session(session *where, session *template) static void Sess_remove_session(session *ses) { + Sess_unhash_session (ses); + if (ses->sort_prev == NULL && ses->sort_next == NULL) /* last session */ Sessions_head = Sessions_tail = NULL; @@ -985,7 +987,6 @@ void Sess_session_denied(int ses) Sessions[ses].mbox ); close( Sessions[ses].mbox ); - Sess_unhash_session (&Sessions[ses]); Sess_remove_session (&Sessions[ses]); Num_sessions--; GlobalStatus.num_sessions = Num_sessions; @@ -1964,7 +1965,6 @@ static void Sess_handle_kill( message_link *mess_link ) /* delete session ses */ if( Is_op_session( Sessions[ses].status ) ) Alarm( EXIT, "Sess_handle_kill: killing unkilled session bug!\n"); - Sess_unhash_session (&Sessions[ses]); Sess_remove_session (&Sessions[ses]); Num_sessions--; GlobalStatus.num_sessions = Num_sessions; Patch 2: -------- Moved Sess_unash_session() calls into Sess_remove_session(). Sess_unhash_session() is always called immediately before Sess_remove_session(). I simplified the call points by moving the calls to the first line of Sess_remove_session(). diff --git a/daemon/session.c b/daemon/session.c index 03af612..b7b5ba9 100644 --- a/daemon/session.c +++ b/daemon/session.c @@ -228,6 +228,8 @@ static int Sess_insert_new_session(session *where, session *template) static void Sess_remove_session(session *ses) { + Sess_unhash_session (ses); + if (ses->sort_prev == NULL && ses->sort_next == NULL) /* last session */ Sessions_head = Sessions_tail = NULL; @@ -985,7 +987,6 @@ void Sess_session_denied(int ses) Sessions[ses].mbox ); close( Sessions[ses].mbox ); - Sess_unhash_session (&Sessions[ses]); Sess_remove_session (&Sessions[ses]); Num_sessions--; GlobalStatus.num_sessions = Num_sessions; @@ -1964,7 +1965,6 @@ static void Sess_handle_kill( message_link *mess_link ) /* delete session ses */ if( Is_op_session( Sessions[ses].status ) ) Alarm( EXIT, "Sess_handle_kill: killing unkilled session bug!\n"); - Sess_unhash_session (&Sessions[ses]); Sess_remove_session (&Sessions[ses]); Num_sessions--; GlobalStatus.num_sessions = Num_sessions; Patch 3: -------- Removed unused mbox parameter from Sess_validate_read_header(). diff --git a/daemon/session.c b/daemon/session.c index b7b5ba9..df3f42e 100644 --- a/daemon/session.c +++ b/daemon/session.c @@ -1057,7 +1057,7 @@ void Sess_session_authorized(int ses) Sessions[ses].mbox ); } -static int Sess_validate_read_header( mailbox mbox, int ses, int head_size, message_header *head_ptr) +static int Sess_validate_read_header( int ses, int head_size, message_header *head_ptr) { char private_name[MAX_PRIVATE_NAME+1]; char proc_name[MAX_PROC_NAME]; @@ -1216,7 +1216,7 @@ static struct msghdr msgh; /* Validate all fields */ Alarm( SESSION, "Sess_read: Message has type field 0x%x\n", head_ptr->type); - ret = Sess_validate_read_header( mbox, ses, head_size, head_ptr); + ret = Sess_validate_read_header( ses, head_size, head_ptr); if (ret < 0 ) { /* invalid header */ Patch 4: -------- Replaced instances of Sessions[ses].mbox with mbox arg in Sess_read(). Sess_read supplies the file descriptor via the mbox argument. There is no need to redundantly use Sessions[ses].mbox after fetching the session index via Sess_get_session_index(mbox). diff --git a/daemon/session.c b/daemon/session.c index df3f42e..f7462a0 100644 --- a/daemon/session.c +++ b/daemon/session.c @@ -1187,24 +1187,24 @@ static struct msghdr msgh; } else if (ret > 0 ) { Sessions[ses].read.cur_byte += ret; ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); return; } else { /* error reading */ if ( (ret == -1) && ( (sock_errno == EINTR) || (sock_errno == EAGAIN) || (sock_errno == EWOULDBLOCK) ) ) { ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); return; } Alarm( SESSION, "Sess_read: failed receiving header on session %d: ret %d: error: %s \n", mbox, ret, sock_strerror(sock_errno) ); Sess_kill( mbox ); ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); return; } /* When we get here we have a complete header */ ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); /* Fliping message header to my form if needed */ if( !Same_endian( head_ptr->type ) ) @@ -1273,26 +1273,26 @@ static struct msghdr msgh; Sessions[ses].read.cur_byte += ret; Sessions[ses].read.total_bytes += ret; ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); return; } else { if ( (ret == -1) && ((sock_errno == EINTR) || (sock_errno == EAGAIN) || (sock_errno == EWOULDBLOCK)) ) { ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); return; } Alarm( SESSION, "Sess_read: failed receiving message on session %d, ret is %d: error: %s\n", mbox, ret, sock_strerror(sock_errno) ); Alarm( SESSION, "Sess_read: failed recv msg more info: len read: %d, remain: %d, to_read: %d, pkt_index: %d, b_index: %d, scat_nums: %d\n",Sessions[ses].read.total_bytes, remain, to_read, packet_index, byte_index, scat->num_elements ); Sess_kill( mbox ); ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); return; } } /* We now have a complete message */ ioctl_cmd = 0; - ioctl( Sessions[ses].mbox, FIONBIO, &ioctl_cmd); + ioctl( mbox, FIONBIO, &ioctl_cmd); /* reset active read_mess to empty */ Message_reset_current_location(&(Sessions[ses].read)); @@ -1417,7 +1417,7 @@ static struct msghdr msgh; * closed socket ourselves and calling Sess_kill(), then the session * is in the wrong state and we will crash when we try to finish delivery. */ - Log_sess_disconnect( Sessions[ses].mbox, &Sessions[ses].addr, Sessions[ses].name, + Log_sess_disconnect( mbox, &Sessions[ses].addr, Sessions[ses].name, Sessions[ses].num_mess ); Patch 5: -------- Fixed bug where old data could be written to newly assigned session. The socket descriptor for a session was being closed before the session itself was removed. That made it a possible for a new session to be created and assigned the old file descriptor before the previous session was removed. That would cause any outstanding messages for the previous session to be delivered to the new session. The solution is to close the session file descriptor when the session is removed, not before. diff --git a/daemon/session.c b/daemon/session.c index f7462a0..7a783fc 100644 --- a/daemon/session.c +++ b/daemon/session.c @@ -126,6 +126,39 @@ static void Sess_deliver_reject( message_obj *msg ); static void Sess_create_reject_message ( message_obj *msg ); static int Sess_get_p2p_dests( int num_groups, char groups[][MAX_GROUP_NAME], char dests[][MAX_GROUP_NAME] ); +/* 2017-01-03 dfs + * + * Make sure there is no leftover data that would remain for a new + * session when the session object is recycled. + */ +static void Sess_clear_session(session *ses) { + E_dequeue_all(ses->mbox); + + while(ses->num_mess > 0) { + message_link* mess_link = ses->first; + ses->first = ses->first->next; + Sess_dispose_message(mess_link); + ses->num_mess--; + } + + ses->first = ses->last = NULL; + + if(ses->read_mess != NULL) { + Message_dispose_message(ses->read_mess); + ses->read_mess = NULL; + } + + Message_reset_current_location(&(ses->read)); + Message_reset_current_location(&(ses->write)); + ses->read.in_mess_head = 1; + + ses->status = Clear_op_session(ses->status); + + E_detach_fd(ses->mbox, READ_FD); + E_detach_fd(ses->mbox, EXCEPT_FD); + E_detach_fd(ses->mbox, WRITE_FD); +} + static void Sess_free_session(session *ses) { ses->sort_next = Sessions_free; @@ -136,6 +169,8 @@ static void Sess_init_sessions(void) { int i; + memset(Sessions, 0, sizeof(Sessions)); + for (i = 0; i < SESSION_FD_HASH_SIZE; ++i) Sessions_hash_head[i] = NULL; @@ -252,6 +287,10 @@ static void Sess_remove_session(session *ses) ses->sort_prev = NULL; ses->sort_next = NULL; Sess_free_session(ses); + + close( ses->mbox ); + /* Make sure we do not alias a future file descriptor assigment. */ + ses->mbox = -1; } int Sess_get_session_index(int mbox) @@ -985,7 +1024,6 @@ void Sess_session_denied(int ses) Alarm( SESSION, "Sess_session_denied: Authorization denied for %s on mailbox %d\n", Sessions[ses].name, Sessions[ses].mbox ); - close( Sessions[ses].mbox ); Sess_remove_session (&Sessions[ses]); Num_sessions--; @@ -1420,23 +1458,8 @@ static struct msghdr msgh; Log_sess_disconnect( mbox, &Sessions[ses].addr, Sessions[ses].name, Sessions[ses].num_mess ); - /* clear his structure */ - while( Sessions[ses].num_mess > 0 ) - { - mess_link = Sessions[ses].first; - Sessions[ses].first = Sessions[ses].first->next; - Sess_dispose_message( mess_link ); - Sessions[ses].num_mess--; - } + Sess_clear_session(&Sessions[ses]); - /* close the mailbox and mark it unoperational */ - E_dequeue( Sess_badger_TO, mbox, NULL ); - E_detach_fd( mbox, READ_FD ); - E_detach_fd( mbox, EXCEPT_FD ); - E_detach_fd( mbox, WRITE_FD ); - close( mbox ); - /* the mailbox is closed but the entry still points to it */ - Sessions[ses].status = Clear_op_session( Sessions[ses].status ); Alarm( SESSION, "Sess_read: disconnecting session %s ( mailbox %d )\n",Sessions[ses].name, mbox ); } @@ -1789,30 +1812,9 @@ static void Sess_kill( mailbox mbox ) Log_sess_disconnect( Sessions[ses].mbox, &Sessions[ses].addr, Sessions[ses].name, Sessions[ses].num_mess ); - - /* clear his structure */ - while( Sessions[ses].num_mess > 0 ) - { - mess_link = Sessions[ses].first; - Sessions[ses].first = Sessions[ses].first->next; - Sess_dispose_message( mess_link ); - Sessions[ses].num_mess--; - } - /* reset active read_mess to empty */ - Message_reset_current_location(&(Sessions[ses].read)); - Sessions[ses].read.in_mess_head = 1; - if (Sessions[ses].read_mess != NULL) - Message_dispose_message( Sessions[ses].read_mess ); - Sessions[ses].read_mess = NULL; - /* close the mailbox and mark it unoperational */ - E_dequeue( Sess_badger_TO, mbox, NULL ); - E_detach_fd( mbox, READ_FD ); - E_detach_fd( mbox, EXCEPT_FD ); - E_detach_fd( mbox, WRITE_FD ); - close(mbox); - /* the mailbox is closed but the entry still points to it */ - Sessions[ses].status = Clear_op_session( Sessions[ses].status ); + Sess_clear_session(&Sessions[ses]); + Alarm( SESSION, "Sess_kill: killing session %s ( mailbox %d )\n",Sessions[ses].name, mbox ); } diff --git a/libspread-util/include/spu_events.h b/libspread-util/include/spu_events.h index 866d72e..8b885db 100644 --- a/libspread-util/include/spu_events.h +++ b/libspread-util/include/spu_events.h @@ -80,6 +80,7 @@ int E_queue( void (* func)( int code, void *data ), int code, void *data, sp_time delta_time ); int E_in_queue( void (* func)( int code, void *data ), int code, void *data ); +int E_dequeue_all( int code ); /* Note: This does not dispose/free the data pointed at by the void *data pointer */ int E_dequeue( void (* func)( int code, void *data ), int code, diff --git a/libspread-util/src/events.c b/libspread-util/src/events.c index 2917a57..0d451f1 100644 --- a/libspread-util/src/events.c +++ b/libspread-util/src/events.c @@ -344,6 +344,43 @@ int E_queue( void (* func)( int code, void *data ), int code, void *data, return( 0 ); } +int E_dequeue_all( int code ) +{ + int found = 0; + time_event *t_pre = Time_queue; + time_event *t_ptr = Time_queue; + + while ( t_ptr != NULL ) + { + time_event *next = t_ptr->next; + + if( t_ptr->code == code ) + { + ++found; + + if(t_ptr == Time_queue) { + Time_queue = next; + } + + if(t_ptr == t_pre) { + t_pre = next; + } else { + t_pre->next = next; + } + + dispose( t_ptr ); + } else { + t_pre = t_ptr; + } + + t_ptr = next; + } + + Alarmp( SPLOG_INFO, EVENTS, "E_dequeue_all: dequeued %d events for code %d\n", found, code ); + + return found; +} + int E_dequeue( void (* func)( int code, void *data ), int code, void *data ) {