Re: [Tarantool-patches] [PATCH] wal: introduce limits on simultaneous writes

[Serge Petrenko via Tarantool-patches <tarantool-patches@dev.tarantool.org>]

31.01.2021 20:13, Vladislav Shpilevoy пишет:
> Hi! Thanks for the patch!

Thanks for the review!

Please find my answers inline and the updated patch in v2 letter.
Some of your comments got obsoleted by the new version, so I left
them with no answer.

> On 27.01.2021 11:11, Serge Petrenko wrote:
>> Since the introduction of asynchronous commit, which doesn't wait for a
>> WAL write to succeed, it's quite easy to clog WAL with huge amounts
>> write requests. For now, it's only possible from an applier, since it's
>> the only user of async commit at the moment.
>>
>> Imagine such a situation: there are 2 servers, a master and a replica,
>> and the replica is down for some period of time. While the replica is
>> down, the master serves requests at a reasonable pace, possibly close to
>> its WAL throughput limit. Once the replica reconnects, it has to receive
>> all the data master has piled up. Now there's no limit in speed at which
>> master sends the data to replica, and there's no limit at which
>> replica's applier submits corresponding write requests to WAL. This
>> leads to a situation when replica's WAL is never in time to serve the
>> requests and the amount of pending requests is constantly growing.
>>
>> To ameliorate such behavior, we need to introduce some limit on
>> not-yet-finished WAL write requests. This is what this commit is trying
>> to do.
>> Two new counters are added to wal writer: queue_size (in bytes) and
>> queue_len (in wal messages) together with configuration settings:
>> `wal_queue_max_size` and `wal_queue_max_len`.
>> Size and length are increased on every new submitted request, and are
>> decreased once the tx receives a confirmation that a specific request
>> was written.
>>
>> Once size or len reach their maximum values, new write requests are
>> blocked (even for async writes) until the queue gets some free space.
>>
>> The size limit isn't strict, i.e. if there's at least one free byte, the
>> whole write request will be added.
>>
>> Part of #5536
>>
>> @TarantoolBot document
>> Title: new configuration options: 'wal_queue_max_size', 'wal_queue_max_len'
>>
>> `wal_queue_max_size` and `wal_queue_max_len` put a limit on the amount
>> of concurrent write requests submitted to WAL.
>> `wal_queue_max_size` is measured in number of bytes to be written (0
>> means unlimited), and `wal_queue_max_len` is measured in number of wal
>> messages (correlates to number of bytes / 1024), 0 meaning unlimited.
>> These options only affect replica behaviour at the moment, and default
>> to 0. They limit the pace at which replica reads new transactions from
>> master.
> The explanation above was brilliant. Could it be somehow fit into the doc
> request, in some a little simplified form? The reason is I want users to know
> when they need to touch these settings. Otherwise they may face the problem,
> and not even realize it is about WAL queue.

Ok, no problem.

> See 9 comments below. One of them is a discussion proposal.
>
>> ---
>> https://github.com/tarantool/tarantool/tree/sp/gh-5536-replica-oom
>> https://github.com/tarantool/tarantool/issues/5536
>>> diff --git a/src/box/wal.c b/src/box/wal.c
>> index 937d47ba9..e38ee8a8e 100644
>> --- a/src/box/wal.c
>> +++ b/src/box/wal.c
>> @@ -166,6 +177,7 @@ struct wal_writer
>>   	 * Used for replication relays.
>>   	 */
>>   	struct rlist watchers;
>> +	struct rlist waiters;
> 1. Worth adding a comment what are the objects in this list, and why
> is it needed.

Done.

>
>>   };
>>   
>>   struct wal_msg {
>> @@ -183,6 +195,29 @@ struct wal_msg {
>>   	struct vclock vclock;
>>   };
>>   
>> +/**
>> + * Possible wal waiter states. There is no "SUCCESS" since the waiter decides
>> + * whether it's succeeded or not on its own.
>> + */
>> +enum wal_waiter_state {
>> +	WAL_WAITER_ROLLBACK = -1,
>> +	WAL_WAITER_PENDING = 0,
>> +};
>> +
>> +/**
>> + * A journal entry waiting for the WAL queue to empty before submitting a write
>> + * request.
>> + */
>> +struct wal_waiter {
>> +	/* The waiting fiber. */
>> +	struct fiber *fiber;
>> +	/* The pending entry. Used for cascading rollback. */
>> +	struct journal_entry *entry;
>> +	enum wal_waiter_state state;
> 2. I suggest to use /** for comments and add a comment to the
> 'state' member. Because we add a comment to each struct member,
> AFAIU almost as a part of our code style. Struct looks cleaner
> then.

Yes, sure. Sorry for the misprints.

>> +	/* Link in waiter list. */
>> +	struct rlist in_list;
>> +};
>> +
>>   /**
>>    * Vinyl metadata log writer.
>>    */
>> @@ -332,6 +367,16 @@ tx_complete_rollback(void)
>>   			      fifo) != writer->last_entry)
>>   		return;
>>   	stailq_reverse(&writer->rollback);
>> +	/*
>> +	 * Every waiting entry came after any of the successfully submitted
>> +	 * entries, so it must be rolled back first to preserve correct order.
>> +	 */
>> +	struct wal_waiter *waiter;
>> +	rlist_foreach_entry(waiter, &writer->waiters, in_list) {
>> +		stailq_add_entry(&writer->rollback, waiter->entry, fifo);
>> +		waiter->state = WAL_WAITER_ROLLBACK;
>> +		fiber_wakeup(waiter->fiber);
>> +	}
>>   	tx_schedule_queue(&writer->rollback);
>>   	/* TX-thread can try sending transactions to WAL again. */
>>   	stailq_create(&writer->rollback);
>> @@ -343,6 +388,16 @@ tx_complete_rollback(void)
>>   	cpipe_push(&writer->wal_pipe, &msg);
>>   }
>>   
>> +static void
>> +wal_wakeup_waiters()
>> +{
>> +	struct wal_writer *writer = &wal_writer_singleton;
>> +	struct wal_waiter *waiter;
>> +	rlist_foreach_entry(waiter, &writer->waiters, in_list)
>> +		fiber_wakeup(waiter->fiber);
>> +}
>> +
>> +
> 3. Extra empty line.


>>   /**
>>    * Complete execution of a batch of WAL write requests:
>>    * schedule all committed requests, and, should there
>> @@ -368,7 +423,15 @@ tx_complete_batch(struct cmsg *msg)
>>   	/* Update the tx vclock to the latest written by wal. */
>>   	vclock_copy(&replicaset.vclock, &batch->vclock);
>>   	tx_schedule_queue(&batch->commit);
>> +	writer->queue_len--;
>> +	writer->queue_size -= batch->approx_len;
>>   	mempool_free(&writer->msg_pool, container_of(msg, struct wal_msg, base));
>> +	/*
>> +	 * Do not wake up waiters if we see there's a rollback planned.
>> +	 * We'll handle them together with other rolled back entries.
>> +	 */
>> +	if (stailq_empty(&writer->rollback))
>> +		wal_wakeup_waiters();
> 4. You can wake them up always. You don't do that only because
> tx_complete_rollback() didn't clear the queue a few lines above, right?
> But it can clear the waiters queue. Because if you set WAL_WAITER_ROLLBACK
> status for a waiter, it can't stay in the queue anymore - in wal_wait_queue()
> you exit when see WAL_WAITER_ROLLBACK. It means it is fine to drop entries
> from the queue which entered a terminal state right away.
>
> You can rlist_create() the entire queue in tx_complete_rollback() and
> remove this 'if (stailq_empty(&writer->rollback))' check. Waiter list will
> be empty in case of rollback. This will optmize the most common case.
>
>>   }
>>   
>>   /**
>> @@ -765,6 +834,27 @@ wal_set_checkpoint_threshold(int64_t threshold)
>>   	fiber_set_cancellable(cancellable);
>>   }
>>   
>> +static inline bool
>> +wal_queue_is_full(void);
> 5. Why do you declare and implement it separately? Does its
> definition need any functions not known here?

>> @@ -1218,6 +1308,40 @@ wal_writer_f(va_list ap)
>>   	return 0;
>>   }
>>   
>> +static inline bool
>> +wal_queue_is_full(void)
>> +{
>> +	struct wal_writer *writer = &wal_writer_singleton;
>> +	return (writer->queue_max_len > 0 &&
>> +		writer->queue_len >= writer->queue_max_len) ||
>> +	       (writer->queue_max_size > 0 &&
>> +		writer->queue_size >= writer->queue_max_size);
>> +}> @@ -1226,6 +1350,7 @@ static int
>>   wal_write_async(struct journal *journal, struct journal_entry *entry)
>>   {
>>   	struct wal_writer *writer = (struct wal_writer *) journal;
>> +	int rc = -1;
> 6. Maybe better set it to JOURNAL_RC_ERROR explicitly.


>>   	ERROR_INJECT(ERRINJ_WAL_IO, {
>>   		goto fail;
>> @@ -1245,6 +1370,11 @@ wal_write_async(struct journal *journal, struct journal_entry *entry)
>>   		goto fail;
>>   	}
>>   
>> +	if (wal_wait_queue(entry) != 0) {
>> +		rc = JOURNAL_RC_ROLLBACK;
>> +		goto fail;
>> +	
> 7. There is a big flaw in this approach: wal_write_async() became not
> async. It now can yield. Even if you would add a flag, it still
> would mean the **async** function can **yield**.
>
> This would be fine if we would need to fix it urgently and then cleanup
> it later as a follow-up ticket. But it is not the case AFAIU, so we
> should find a better way to block the applier. And keep thinking
> about async transactions in userspace in future.
>
> Another issue I see - it affects blocking transactions too, and
> AFAIS increases their latency. Consider how it worked before:
>
> 	- During event loop iteration the transactions stacked
> 	  into one or more wal_msg objects;
>
> 	- Some of the wal_msgs could be flushed to WAL thread's
> 	  queue.
>
> 	- In the end of the event loop all of them are flushed
>            always.
>
> 	- Cbus in WAL thread picks up one or more wal_msg objects,
> 	  and serves them. It can pick more than one wal_msg, as
>            well as tx can send more than one during one event loop
>            iteration.
>
> Here we had batching at 2 levels: inside wal_msg (in wal code),
> and wal_msg themselves (in cbus code).
>
> Now the first step is changed: you stop stacking more wal_msg
> objects when one of the limits is reached. If queue size is reached,
> you stop stacking even one wal_msg entries. Even for synchronous
> commits. Therefore some of them won't fit into the current event
> loop iteration.
>
> This, as I suspect, will slow down normal userspace transactions
> if queues are limited to sane values and fiber count is huge.
>
> I don't yet see a golden solution which would make it look perfect,
> and not affect performance at all, but have some options to discuss.
> Or which could inspire you on some better solution.
>
> Also everything I said above could be bullshit. Better double check it.
>
> **Option 1**
>
> Stupid, and I don't know how to implement without affecting blocking
> transactions. But maybe the idea could be improved to something working.
>
> The idea is if the queue is full, return some kind of a special
> error without waiting. If you get the error, you can either rollback
> or retry using blocking commit.
>
> Issues here:
>
> 	- txn_commit() won't work for applier, because 1) it assumes
> 	  lsns are not known, 2) it acks the limbo;
>
> 	- You can't simply retry txn_commit() or txn_commit_async()
> 	  because they do txn_prepare() and build the entry. I
> 	  don't know how to extract these actions accurately.
>
> **Option 2**
>
> Idea is to introduce API for WAL module to check if the queue is full.
> It would be used by applier and if it sees the queue is full, it
> uses blocking commit.
>
> Or we could add a method like wal_wait_queue(). Because txn_commit(),
> as I said above, won't work with applier.
>
> Issues:
>
> 	- Breaks encapsulation of journal API - you would need to access
> 	  WAL API in applier directly.
>
> **Option 3**, which I like most so far.
>
> Move your code up to the journal API level. To journal.h. We would
> have these methods:
>
> 	journal_queue_is_full()
> 	journal_queue_wait()
>
> - Applier would check if the queue is full, and wait if it is.
>
> - In future box.commit({is_async}) will check if full, and
> return an error.
>
> - journal_write() and journal_write_async() wouldn't check
> if the queue is full. Only increase queue size and len. Because
> they are used box.commit() which does not care about queue limits.
> But they check if the queue is empty. And if it is not, they
> **take all the waiters* alongside. Piggyback them. Flush the queue.
> At least journal_write() will do this.
>
> You must flush the queue then, because if you don't, the queue won't
> disappear anyway. It will still wait and occupy memory. This means if
> you have a blocking commit, it is better to flush the queue. This is
> also needed to preserve the order. If some waiters were there before
> this box.commit(), they should go earlier.
>
> - journal_async_complete() decreases queue_len, queue_size, and
> wakes up a next waiter, if there are any. Or rather wakes up
> exactly as many waiters as necessary to fill the queue again. To
> avoid spurious wakeups.
>
> **Option 4**
>
> Not care about my proposals, and go for the current patch after
> a few amendments according to the other comments. Which I like the
> least, but Kirill may decide to force it.

Thanks for your suggestions!
I took option 3 and it turned out quite well in my opinion.
The patch became considerably simpler.
Please find v2 in your inbox.

>
>> +
>>   	struct wal_msg *batch;
>>   	if (!stailq_empty(&writer->wal_pipe.input) &&
>>   	    (batch = wal_msg(stailq_first_entry(&writer->wal_pipe.input,
>> @@ -1259,6 +1389,7 @@ wal_write_async(struct journal *journal, struct journal_entry *entry)
>>   			goto fail;
>>   		}
>>   		wal_msg_create(batch);
>> +		writer->queue_len++;
> 8. Why don't you increase the queue length always? The request did
> go to the current batch, but still it is a request waiting to be written.
> So it is in the queue. Queue_size, on the other hand, is increased always.
> If we would add monitoring to this, a user would see how queue size is
> growing, and queue len is not. Looks scary.

That's because queue length is measured in wal_msgs rather than in write 
requests.
So that when queue length limit is 1 you may have a single WAL message 
travelling
between wal and tx at all times. But this message may carry several 
write requests.
I'm not sure whether we should leave this limit at all. Maybe a size 
limit alone would
be enough.

Anyway, I changed this limit to count journal_entries (transactions). 
Still not sure whether
it's needed.

>>   		/*
>>   		 * Sic: first add a request, then push the batch,
>>   		 * since cpipe_push() may pass the batch to WAL
>> @@ -1274,6 +1405,7 @@ wal_write_async(struct journal *journal, struct journal_entry *entry)
>>   	 */
>>   	writer->last_entry = entry;
>>   	batch->approx_len += entry->approx_len;
>> +	writer->queue_size += entry->approx_len;
>>   	writer->wal_pipe.n_input += entry->n_rows * XROW_IOVMAX;
>>   #ifndef NDEBUG
>>   	++errinj(ERRINJ_WAL_WRITE_COUNT, ERRINJ_INT)->iparam;
>> @@ -1283,7 +1415,7 @@ wal_write_async(struct journal *journal, struct journal_entry *entry)
>>   
>>   fail:
>>   	entry->res = -1;
> 9. Maybe better use entry->res for returning the reason? So it would work
> just like txn->signature. The code would become a bit simpler and more
> consistent with the similar existing hack.
>
>> -	return -1;
>> +	return rc;
>>   }

-- 
Serge Petrenko