Re: [Tarantool-patches] [tarantool-patches] Re: [PATCH v2 4/4] replication: use wal memory buffer to fetch rows

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

See 5 comments below.

>> diff --git a/src/box/relay.cc b/src/box/relay.cc
>> index 21674119d..1e65d6d56 100644
>> --- a/src/box/relay.cc
>> +++ b/src/box/relay.cc
>> @@ -161,9 +163,9 @@ relay_new(struct replica *replica)
>>  	}
>>  	relay->replica = replica;
>>  	relay->last_row_time = ev_monotonic_now(loop());
>> -	fiber_cond_create(&relay->reader_cond);
>>  	diag_create(&relay->diag);
>>  	relay->state = RELAY_OFF;
>> +	relay->r = NULL;
> 
> 1. Why? 'relay' is allocated using 'calloc' a few lines
> above, 'r' is NULL already.
> 

1. Why did you ignore that comment?

>> +	relay->wal_dir = strdup(cfg_gets("wal_dir"));
>> +	if (relay->wal_dir == NULL) {
>> +		diag_set(OutOfMemory, strlen(cfg_gets("wal_dir")),
> 
> 10. I know, I am a fucking bore, but 'strlen() + 1', for terminating
> zero :)
> 

2. Why did you ignore that comment?

>> +			 "runtime", "wal_dir");
> 
> 11. There is a strict rule what arguments we pass to diag_raise -
> it is size, allocator function, variable name. Allocator function
> here is 'strdup', not 'runtime'.
> 

3. Why did you ignore that comment?

>> +	struct fiber_cond xrow_buf_cond;
>>  };
>>  
>>  struct wal_msg {
>> @@ -1131,6 +1134,7 @@ wal_writer_f(va_list ap)
>>  	(void) ap;
>>  	struct wal_writer *writer = &wal_writer_singleton;
>>  	xrow_buf_create(&writer->xrow_buf);
>> +	fiber_cond_create(&writer->xrow_buf_cond);
> 
> 18. Where do you call fiber_cond_destroy()?
> 

4. Why did you ignore that comment?

>> +
>>  #if defined(__cplusplus)
>>  } /* extern "C" */
>>  #endif /* defined(__cplusplus) */
>> diff --git a/src/lib/core/cbus.c b/src/lib/core/cbus.c
>> index b3b1280e7..b7e6d769b 100644
>> --- a/src/lib/core/cbus.c
>> +++ b/src/lib/core/cbus.c
>> @@ -284,6 +284,9 @@ cpipe_flush_cb(ev_loop *loop, struct ev_async *watcher, int events)
>>  	/* Trigger task processing when the queue becomes non-empty. */
>>  	bool output_was_empty;
>>  
>> +	int old_cancel_state;
>> +	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancel_state);
> 
> 28. Why?
> 

5. Why did you ignore that comment?

>> +
>>  	tt_pthread_mutex_lock(&endpoint->mutex);
>>  	output_was_empty = stailq_empty(&endpoint->output);
>>  	/** Flush input */
>> @@ -297,6 +300,7 @@ cpipe_flush_cb(ev_loop *loop, struct ev_async *watcher, int events)
>>  
>>  		ev_async_send(endpoint->consumer, &endpoint->async);
>>  	}
>> +	pthread_setcancelstate(old_cancel_state, NULL);
>>  }
> 
> 29. I will review the tests later, when I will fully understand the
> code. But so few tests for such a serious feature looks not enough.
> Could you test it more rigorous?
> 

6. So have you succeed in that hacking? I am mostly
interested in tests about slow replication with fast
WAL writing.