Add manual locking to ASYNCIO for transacted writes

Previously, e. g. it was possible that newlines are separated from the
printed line in `logger_file`.

src/base/system.c

@@ -164,8 +164,10 @@ static void logger_debugger(const char *line, void *user)
 static void logger_file(const char *line, void *user)
 {
 	ASYNCIO *logfile = (ASYNCIO *)user;
-	aio_write(logfile, line, strlen(line));
-	aio_write_newline(logfile);
+	aio_lock(logfile);
+	aio_write_unlocked(logfile, line, strlen(line));
+	aio_write_newline_unlocked(logfile);
+	aio_unlock(logfile);
 }
 
 static void logger_stdout_finish(void *user)
@@ -624,10 +626,20 @@ static unsigned int next_buffer_size(unsigned int cur_size, unsigned int need_si
 	return cur_size;
 }
 
-void aio_write(ASYNCIO *aio, const void *buffer, unsigned size)
+void aio_lock(ASYNCIO *aio)
+{
+	lock_wait(aio->lock);
+}
+
+void aio_unlock(ASYNCIO *aio)
+{
+	lock_unlock(aio->lock);
+	sphore_signal(&aio->sphore);
+}
+
+void aio_write_unlocked(ASYNCIO *aio, const void *buffer, unsigned size)
 {
 	unsigned int remaining;
-	lock_wait(aio->lock);
 	remaining = aio->buffer_size - buffer_len(aio);
 
 	// Don't allow full queue to distinguish between empty and full queue.
@@ -680,17 +692,29 @@ void aio_write(ASYNCIO *aio, const void *buffer, unsigned size)
 		aio->read_pos = 0;
 		aio->write_pos = next_len;
 	}
-	lock_unlock(aio->lock);
-	sphore_signal(&aio->sphore);
+}
+
+void aio_write(ASYNCIO *aio, const void *buffer, unsigned size)
+{
+	aio_lock(aio);
+	aio_write_unlocked(aio, buffer, size);
+	aio_unlock(aio);
+}
+
+void aio_write_newline_unlocked(ASYNCIO *aio)
+{
+#if defined(CONF_FAMILY_WINDOWS)
+	aio_write_unlocked(aio, "\r\n", 2);
+#else
+	aio_write_unlocked(aio, "\n", 1);
+#endif
 }
 
 void aio_write_newline(ASYNCIO *aio)
 {
-#if defined(CONF_FAMILY_WINDOWS)
-	aio_write(aio, "\r\n", 2);
-#else
-	aio_write(aio, "\n", 1);
-#endif
+	aio_lock(aio);
+	aio_write_newline_unlocked(aio);
+	aio_unlock(aio);
 }
 
 int aio_error(ASYNCIO *aio)

src/base/system.h

@@ -382,6 +382,26 @@ typedef struct ASYNCIO ASYNCIO;
 */
 ASYNCIO *aio_new(IOHANDLE io);
 
+/*
+	Function: aio_lock
+		Locks the ASYNCIO structure so it can't be written into by
+		other threads.
+
+	Parameters:
+		aio - Handle to the file.
+*/
+void aio_lock(ASYNCIO *aio);
+
+/*
+	Function: aio_unlock
+		Unlocks the ASYNCIO structure after finishing the contiguous
+		write.
+
+	Parameters:
+		aio - Handle to the file.
+*/
+void aio_unlock(ASYNCIO *aio);
+
 /*
 	Function: aio_write
 		Queues a chunk of data for writing.
@@ -404,6 +424,30 @@ void aio_write(ASYNCIO *aio, const void *buffer, unsigned size);
 */
 void aio_write_newline(ASYNCIO *aio);
 
+/*
+	Function: aio_write_unlocked
+		Queues a chunk of data for writing. The ASYNCIO struct must be
+		locked using `aio_lock` first.
+
+	Parameters:
+		aio - Handle to the file.
+		buffer - Pointer to the data that should be written.
+		size - Number of bytes to write.
+
+*/
+void aio_write_unlocked(ASYNCIO *aio, const void *buffer, unsigned size);
+
+/*
+	Function: aio_write_newline_unlocked
+		Queues a newline for writing. The ASYNCIO struct must be locked
+		using `aio_lock` first.
+
+	Parameters:
+		aio - Handle to the file.
+
+*/
+void aio_write_newline_unlocked(ASYNCIO *aio);
+
 /*
 	Function: aio_error
 		Checks whether errors have occured during the asynchronous

src/test/aio.cpp

@@ -122,3 +122,25 @@ TEST_F(Async, NonDivisor)
 	}
 	Expect(aText);
 }
+
+TEST_F(Async, Transaction)
+{
+	static const int NUM_LETTERS = 13;
+	static const int SIZE = BUF_SIZE / NUM_LETTERS * NUM_LETTERS;
+	char aText[SIZE + 1];
+	for(unsigned i = 0; i < sizeof(aText) - 1; i++)
+	{
+		aText[i] = 'a' + i % NUM_LETTERS;
+	}
+	aText[sizeof(aText) - 1] = 0;
+	for(unsigned i = 0; i < (sizeof(aText) - 1) / NUM_LETTERS; i++)
+	{
+		aio_lock(m_pAio);
+		for(char c = 'a'; c < 'a' + NUM_LETTERS; c++)
+		{
+			aio_write_unlocked(m_pAio, &c, 1);
+		}
+		aio_unlock(m_pAio);
+	}
+	Expect(aText);
+}