Data Logging using IPC on Windows

• Download source code - 12.7 KB

When we need to understand the way a piece of code works, most of us use logging. Most people use file based logging solutions. But when you are developing a product, logging to a file feels like a two step operation:

1. log what happened
2. see in the log what had happened

I never liked that. So I implemented this class for logging to a second application that interactively shows what's going on. This is yet another MMF IPC data transfer class, but with the particularities of a logging tool: it has one operative process that simply collects all the logged data and one or more passive process that generates and writes the log information.

Application logging data 1 ---\
Application logging data 2 --------> MMF ---> Logging collector
Application logging data 3 ---/

I think this logic is faster than most of the other options you have for application logging. And at the same time, it allows you to merge log data from more than one application running on the same computer.

The most important thing here is to set up a well orchestrated set of events and mutex to know when you can read and when you can write and never, never ever overwrite a content that hasn't been read already.

We use just two events for all that log: a read event, that flags when a read operation took place and an avail event that flags when a write operation took place. We also make use of a Memory Mapped File header structure to store shared values like where we are about to read next, or where we are about to write and those kind of things. Finally, because we can't have more than one thread writing data at the same time, we use a mutex to keep any other thread out of the writing method, while writing is taking place.

If we are the reading application, we run a thread that will wait for new data all the time. This is very important because any data that comes to the MMF, we have to process it as fast as possible. We use two kinds of data here but it might be extended to many more kinds of data. The first one is the APPLICATION_LOGGING which is the log itself, and the other one we call it APPLICATION_EOF. We have a block of memory (the memory mapped file) and we have to write on it while we have enough space for doing so. We cannot strip the message in pieces. In case we want to write a new log message and we don't have enough free space, we have to write an EOF and go back to the beginning of the memory area to write our message there. In case the reading thread didn't free up enough space on the beginning of the shared memory, we have to wait for the next read event. Once the reader has read, we check again for enough space, if we have enough space, we go on and write the new message, otherwise we wait for the next read event.

Let's say we use R as the next reading place, W as the next writing place,
X as data written and not read so far and Z as EOF.

We start with all memory available so we just write
|RXXXXXXXXXXW------------------------------------------------------------------------------|

We continue writing until we have no more free space available
|-----------RXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXW-------|

We can write an EOF (Z) and continue in the beginning
|XXXXXXXXW-----------------RXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXZ-------|

This logic looks good enough but there is a catch. In the beginning, we have all the shared memory free and next reading place is the same as the next writing place. What if the writer fills all the memory before the reader can even start reading something? Last writer operation will be writing EOF and go to the beginning of the shared memory. Again, we will have next reading place the same as next writing place. Should we write? Of course no. But how do we know when this is the case? I solve that using a "round mask" for writer and reader. Each time I write an EOF I switch the round mask, so if next writing pointer is the same as next reading pointer and both have the same "round mask", the memory is free and we can write. If next writing pointer is the same as next reading pointer and they have a different "round mask", the memory wasn't being read and we have to wait for the next read event. You can see this logic in log data function:

BOOLEAN data_logging::log_data(PLOG_PACKAGE_DATA data)
{
	if (!m_running) return false;
	UINT16 transfer_size = data->data_size + data->header_size;
	if ((NULL == m_mapFile) || (NULL == m_memory)) {
		error_handler(L"log_data > memory mapped file and/or memory pointer is NULL!");
		goto log_error;
	}

	if (m_data_header->first_element_offset + transfer_size /* this packet */ + 
                           sizeof(MMFILE_HEADER) /* the needed EOF packet */ > MMFILE_SIZE)
	{
		// the package is too big and can't be handled here
		error_handler(L"Package too big. Can't handle packages greater than %u bytes.", 
                      MMFILE_SIZE - sizeof(MMFILE_HEADER) - m_data_header->first_element_offset);
		goto log_error;
	}

	if (WaitForSingleObject(m_log_writers_lock, INFINITE) != WAIT_OBJECT_0)
	{
		// the package is too big and can't be handled here
		error_handler(L"Unable to obtain writer lock - aborting");
		goto log_error;
	}
	UINT32 writeOffset = m_data_header->next_writing_pointer;
	BOOLEAN needToWriteEOFPacket = writeOffset + transfer_size /* this packet */ + 
                    sizeof(MMFILE_HEADER) /* the needed EOF packet */ > MMFILE_SIZE;

	// in case the reader is in the previous round, we have to wait for it to catch up us 
	while (((m_data_header->flags[LOG_FLAG_TEXT] & WRITE_MMF_ROUND_MASK) != 0) != 
                        ((m_data_header->flags[LOG_FLAG_TEXT] & READ_MMF_ROUND_MASK) != 0)) {
		int waitResult = WaitForSingleObject(m_log_read_ev, MAX_WAIT_FOR_WRITE);
		if ((waitResult != WAIT_OBJECT_0) && (((m_data_header->flags[LOG_FLAG_TEXT] & 
                      WRITE_MMF_ROUND_MASK) != 0) != ((m_data_header->flags[LOG_FLAG_TEXT] & 
                      READ_MMF_ROUND_MASK) != 0)))
		{
			error_handler(L"log_data > Write wait operation failed - waitResult = %u", 
                                      waitResult);
			ReleaseMutex(m_log_writers_lock);
			goto log_error;
		}
	}

	// manage the EOF write as a full stand-alone package so in case we need to write more 
    // than MMFILE_SIZE / 2 we can wait for the reader to read up to the EOFPacket and then 
    // we will have the whole MMF for our huge package
	if (needToWriteEOFPacket)
	{
		writeEOF();
		m_data_header->next_writing_pointer = writeOffset = 
                                        m_data_header->first_element_offset;
		char logTextFlags = m_data_header->flags[LOG_FLAG_TEXT];
		while (logTextFlags != _InterlockedCompareExchange8
                         ((char*)&m_data_header->flags[LOG_FLAG_TEXT], 
                          logTextFlags ^ WRITE_MMF_ROUND_MASK, logTextFlags))
		{
			logTextFlags = m_data_header->flags[LOG_FLAG_TEXT];
		}
		if (!SetEvent(m_log_avail_ev))
		{
			// unexpected error!
			error_handler(L"log_data > SetEvent(m_log_avail_ev) failed!");
		}
	}

	while (wait_for_read_before_write(writeOffset, transfer_size)) {
		int waitResult = WaitForSingleObject(m_log_read_ev, MAX_WAIT_FOR_WRITE);
		if ((waitResult != WAIT_OBJECT_0) && wait_for_read_before_write
                            (writeOffset, transfer_size))
		{
			error_handler(L"log_data > Write wait operation failed - waitResult = %u", 
                                      waitResult);
			ReleaseMutex(m_log_writers_lock);
			goto log_error;
		}
	}
	writeOffset += (UINT32)write(writeOffset, data->header, data->header_size);
	writeOffset += (UINT32)write(writeOffset, data->data, data->data_size);
	m_data_header->next_writing_pointer = writeOffset;
	ReleaseMutex(m_log_writers_lock);
	if (!SetEvent(m_log_avail_ev))
	{
		// unexpected error!
		error_handler(L"log_data > SetEvent(m_log_avail_ev) (2) failed!");
	}

	return true;
log_error:
	m_running = false;
	return false;
}

Function wait_for_read_before_write checks to see if we have enough free space for our data or we have to wait, considering round mask as well.

BOOLEAN data_logging::wait_for_read_before_write(UINT32 write_offset, UINT16 data_size) 
{
	// The logic here is that it might happen that we are about to write back from where the other 
    // thread is reading, so we have to make sure that we don't override the data that has not being
    // read yet. That explains the unequal sign on both sides. But what if we wrote all the data 
    // until the position where the reading pointer is (in it's previous round)?
	// Using the current check we will miss the possibility that we are in different rounds and we 
    // are about to write over the unread data. In order to avoid that case, we include the 
    // next_reading_pointer as a position to avoid as well.
	// >>>>>>>>>>>>>>>>W------------------------R>>>>>>>>>>>>>>>>>>>//
	BOOLEAN differentRound = ((m_data_header->flags[LOG_FLAG_TEXT] & WRITE_MMF_ROUND_MASK) 
              != 0) != ((m_data_header->flags[LOG_FLAG_TEXT] & READ_MMF_ROUND_MASK) != 0);
	BOOLEAN res = ((write_offset < m_data_header->next_reading_pointer) && 
               (write_offset + data_size >= m_data_header->next_reading_pointer)) || 
		(differentRound && (write_offset == m_data_header->next_reading_pointer));
	return res;
}

Finally, reading thread executes waitForLog function:

DWORD WINAPI data_logging::waitForLog(LPVOID data)
{
	data_logging * self = (data_logging *)data;
	HANDLE waitOn[2];
	waitOn[0] = self->m_log_avail_ev;
	waitOn[1] = self->m_terminate;

	while (true)
	{
		int wait_result = WaitForMultipleObjects(2, waitOn, FALSE, INFINITE);
		if (wait_result == WAIT_OBJECT_0 + 1) // terminate has been triggered
		{
			self->m_running = false;
			break;
		}
		if (wait_result != WAIT_OBJECT_0) // unexpected result! terminate 
                                                  // as well - but report error!
		{
			self->error_handler(L"waitForLog > WaitForMultipleObjects failed - 
                                                           waitResult = %u\n", wait_result);
			goto error_case;
			break;
		}
		// we have data in the MMF, so look for it. Have to finish when both pointers 
                // have the same value and both operations are in the same round.
		while ((self->m_data_header->next_reading_pointer != 
                                   self->m_data_header->next_writing_pointer) || 
			(((self->m_data_header->flags[LOG_FLAG_TEXT] & WRITE_MMF_ROUND_MASK) != 0) != 
                           ((self->m_data_header->flags[LOG_FLAG_TEXT] & READ_MMF_ROUND_MASK) != 0)))
		{
			switch (((PLOG_HEADER)((char*)self->m_memory + self->m_data_header->
                                          next_reading_pointer))->application)
			{
			case APPLICATION_EOF:
			{
				self->m_data_header->next_reading_pointer = self->m_data_header->
                                         first_element_offset;
				char logTextFlag = self->m_data_header->flags[LOG_FLAG_TEXT];
				while (logTextFlag != _InterlockedCompareExchange8((char *)
                                       &self->m_data_header->flags[LOG_FLAG_TEXT], 
                                       logTextFlag ^ READ_MMF_ROUND_MASK, logTextFlag))
				{
					logTextFlag = self->m_data_header->flags[LOG_FLAG_TEXT];
				}
				break;
			}
			case APPLICATION_LOGGING:
			{
				PLOG_MESSAGES msg_header = (PLOG_MESSAGES)((char*)self->m_memory + 
                                         self->m_data_header->next_reading_pointer);
				int message_length = (msg_header->header.size - 
                                                        sizeof(LOG_MESSAGES))/2 - 1;
				wchar_t * message = (wchar_t*)((char*)msg_header + 
                                                                     sizeof(LOG_MESSAGES));
				if (self->m_kind_callbacks.find(APPLICATION_LOGGING) != 
                                                            self->m_kind_callbacks.end())
					((logAvailableCallbackFunction)self->
                                             m_kind_callbacks[APPLICATION_LOGGING])
                                            (msg_header, message, message_length, 
                                            self->m_kind_callbacks_parameter[APPLICATION_LOGGING]);
				self->m_data_header->next_reading_pointer += msg_header->header.size;
				break;
			}
			default:
				self->error_handler(L"waitForLog > Unexpected application code: %u\n", 
                                     ((PLOG_HEADER)((char*)self->m_memory + self->m_data_header->
                                      next_reading_pointer))->application);
				goto error_case;
				break;
			}
			if (!SetEvent(self->m_log_read_ev))
			{
				// unexpected error! log
				self->error_handler(L"waitForLog > Unexpected error, 
                                                SetEvent of m_log_read_ev failed\n");
			}
		}
	}
	return 0;
error_case:
	self->m_running = false;
	return 0;
}

I use it every day and it works flawlessly. I'm sure it can be polished a bit more, but the way it is, it is good enough for the purpose I pretend for it. You can extend it adding more data kinds. The only restriction is that it is a one way communication logic: many writers and just one listener that cannot answer back to them. In case you want to have a two way communication, you cannot do that with this logic, because the writer is set up only when the reader is already listening (the reader creates the MMF). You have to reformulate this logic to have a client-server logic with two memory mapped files (one for reading and the other for writing), which is the topic of the next article where I present a two way IPC I use to communicate JazzVPN service with GUI front end.