Don't open the control socket until after the server socket is bound

This makes it easier for the tests (and supervisor) to guarantee to be
able to connect to the server socket.

Also this patch moves freeing the mirror supervisor into the server
thread.
This commit is contained in:
Alex Young
2012-10-09 17:35:20 +01:00
parent 83eb31aba4
commit 22bea81445
12 changed files with 108 additions and 69 deletions

View File

@@ -210,11 +210,13 @@ file check("flexnbd") =>
gcc_link t.name, t.prerequisites + [LIBCHECK]
end
file check("control") =>
%w{build/tests/check_control.o} + OBJECTS - ["build/main.o"] do |t|
gcc_link t.name, t.prerequisites + [LIBCHECK]
end
(TEST_MODULES- %w{control flexnbd acl client serve readwrite util}).each do |m|
tgt = "build/tests/check_#{m}.o"
maybe_obj_name = "build/#{m}.o"

View File

@@ -54,6 +54,7 @@ struct control * control_create(
control->flexnbd = flexnbd;
control->socket_name = csn;
control->open_signal = self_pipe_create();
control->close_signal = self_pipe_create();
control->mirror_state_mbox = mbox_create();
@@ -75,6 +76,7 @@ void control_destroy( struct control * control )
mbox_destroy( control->mirror_state_mbox );
self_pipe_destroy( control->close_signal );
self_pipe_destroy( control->open_signal );
free( control );
}
@@ -205,10 +207,23 @@ void control_listen(struct control* control)
control->control_fd = open_control_socket( control->socket_name );
}
void control_wait_for_open_signal( struct control * control )
{
fd_set fds;
FD_ZERO( &fds );
self_pipe_fd_set( control->open_signal, &fds );
FATAL_IF_NEGATIVE( select( FD_SETSIZE, &fds, NULL, NULL, NULL ),
"select() failed" );
self_pipe_signal_clear( control->open_signal );
}
void control_serve( struct control * control )
{
NULLCHECK( control );
control_wait_for_open_signal( control );
control_listen( control );
while( control_accept( control ) );
}

View File

@@ -19,6 +19,7 @@ struct control {
pthread_t thread;
struct self_pipe * open_signal;
struct self_pipe * close_signal;
/* This is owned by the control object, and used by a

View File

@@ -235,13 +235,15 @@ int flexnbd_serve( struct flexnbd * flexnbd )
{
NULLCHECK( flexnbd );
int success;
struct self_pipe * open_signal = NULL;
if ( flexnbd->control ){
debug( "Spawning control thread" );
flexnbd_spawn_control( flexnbd );
open_signal = flexnbd->control->open_signal;
}
success = do_serve( flexnbd->serve );
success = do_serve( flexnbd->serve, open_signal );
debug("do_serve success is %d", success );
if ( flexnbd->control ) {

View File

@@ -566,13 +566,6 @@ void * mirror_super_runner( void * serve_uncast )
}
while ( should_retry && !success );
serve->mirror = NULL;
serve->mirror_super = NULL;
server_allow_mirror_start( serve );
mirror_super_destroy( super );
debug( "Mirror supervisor done." );
return NULL;
}

View File

@@ -187,7 +187,6 @@ char * help_help_text = help_help_text_arr;
int do_serve(struct server* params);
void do_read(struct mode_readwrite_params* params);
void do_write(struct mode_readwrite_params* params);
void do_remote_command(char* command, char* mode, int argc, char** argv);

View File

@@ -731,7 +731,7 @@ int server_accept( struct server * params )
if ( 0 < signal_fd && FD_ISSET( signal_fd, &fds ) ){
debug( "Stop signal received." );
server_close_clients( params );
params->success = serve_shutdown_is_graceful( params );
params->success = params->success && serve_shutdown_is_graceful( params );
should_continue = 0;
}
@@ -823,6 +823,7 @@ void serve_wait_for_close( struct server * serve )
*/
void server_control_arrived( struct server *serve )
{
debug( "server_control_arrived" );
NULLCHECK( serve );
if ( !serve->success ) {
@@ -908,6 +909,9 @@ int server_is_mirroring( struct server * serve )
return !!serve->mirror_super;
}
void mirror_super_destroy( struct mirror_super * super );
/* This must only be called with the start_mirror lock held */
void server_abandon_mirror( struct server * serve )
{
@@ -924,6 +928,14 @@ void server_abandon_mirror( struct server * serve )
pthread_t tid = serve->mirror_super->thread;
pthread_join( tid, NULL );
debug( "Mirror thread %p pthread_join returned", tid );
server_allow_mirror_start( serve );
mirror_super_destroy( serve->mirror_super );
serve->mirror = NULL;
serve->mirror_super = NULL;
debug( "Mirror supervisor done." );
}
}
@@ -934,7 +946,7 @@ int server_default_deny( struct server * serve )
}
/** Full lifecycle of the server */
int do_serve(struct server* params)
int do_serve( struct server* params, struct self_pipe * open_signal )
{
NULLCHECK( params );
@@ -942,6 +954,11 @@ int do_serve(struct server* params)
error_set_handler((cleanup_handler*) serve_cleanup, params);
serve_open_server_socket(params);
/* Only signal that we are open for business once the server
socket is open */
if ( NULL != open_signal ) { self_pipe_signal( open_signal ); }
serve_init_allocation_map(params);
serve_accept_loop(params);
success = params->success;

View File

@@ -133,7 +133,7 @@ int server_mirror_can_start( struct server *serve );
void server_unlink( struct server * serve );
int do_serve( struct server * );
int do_serve( struct server *, struct self_pipe * );
struct mode_readwrite_params {
union mysockaddr connect_to;

View File

@@ -9,7 +9,7 @@ struct status * status_create( struct server * serve )
status = xmalloc( sizeof( struct status ) );
status->pid = getpid();
status->has_control = server_is_in_control( serve );
status->has_control = serve->success;
status->is_mirroring = NULL != serve->mirror;
return status;

View File

@@ -97,7 +97,9 @@ class ValgrindKillingExecutor
when "line"
@error.add_line( @text ) if @found
when "error", "stack"
@killer.call( @error )
if @found
@killer.call( @error )
end
when "pid"
@error.pid=@text
end
@@ -326,8 +328,10 @@ module FlexNBD
def serve( file, *acl)
cmd = serve_cmd( file, acl )
run_serve_cmd( cmd )
sleep( 0.2 ) until File.exists?( ctrl )
end
def listen(file, *acl)
run_serve_cmd( listen_cmd( file, acl ) )
end

View File

@@ -9,10 +9,16 @@ module FlexNBD
def initialize( addr, port, err_msg, source_addr=nil, source_port=0 )
timing_out( 2, err_msg ) do
@sock = if source_addr
TCPSocket.new( addr, port, source_addr, source_port )
else
TCPSocket.new( addr, port )
begin
@sock = if source_addr
TCPSocket.new( addr, port, source_addr, source_port )
else
TCPSocket.new( addr, port )
end
rescue Errno::ECONNREFUSED
$stderr.puts "Connection refused, retrying"
sleep(0.2)
retry
end
end
end

View File

@@ -22,7 +22,7 @@ START_TEST( test_gets_has_control )
struct server server;
struct status * status;
server.has_control = 1;
server.success = 1;
status = status_create( &server );
fail_unless( status->has_control == 1, "has_control wasn't copied" );