Broadcast Burst Queue

• Download source - 25.7 KB
• Download demo - 158.9 KB

Introduction

BBQ (acronym for Broadcast Burst Queue, we call it shortly "barbeque") is a small and useful sample library for accelerated message sending via Windows Sockets. It used to be a part of our testing set of communication libraries that we finally decided not to use in Safetica's product. That’s why we are now free to share the code with the rest of you.

Background

Writing any kind of IPC is always a question of requirements, needs and design, and a trade-off between them, of course. So, let’s see what were the main requirements we had for BBQ:

1. One or more multi-threaded scalable servers per PC that are handling "unlimited" number of clients.
2. Communication is one-way only, no replies are required (client ->rver).
3. We don’t want to use any synchronization mechanisms that could lead to deadlocks, both client and server should be able to crash at any time without any need to release locks to keep system stable, etc.
4. We want to use system buffers as much as possible, to achieve maximum performance and responsivity. We want to be able to push large amount of data and the system should be able to handle it without lowering our performance.
5. We want to be able to dynamically change reconnection parameters (wait time and number of reconnection attempts) based upon the current workload.
6. We want the clients and the server to be able to work on different computers. Sent data should be centralised over the network.

You can easily realize that a typical usage for this project is some kind of inter-process client/server logging, and you are right. Since it uses Windows Sockets, BBQ can work in a network where the server and all clients are running on a different computers.

There are a lot of ways how to deal with one server with multiple client connections. The question is: how to force operating system to hold all the other clients’ connections until the current one is properly handled? And how to process these messages as fast as possible and use as much of CPU time as possible to keep other clients waiting no longer than is absolutely necessary? Many useful articles were already discussing this issue, I can recommend you these three, all from the same author - Changs Hu Liu (thanks man:):

Very refreshing and detailed comparision of all possible I/O approaches is available here, I highly recommend you to read it, illustrative pictures are included (thanks to Thomas Bleeker!):

You can find a lot of code samples at:

To achieve what we wanted, we used these various approaches in BBQ:

1. Overlapped I/O operations

   Simply said, overlapped operations are sort of asynchronous I/O calls that allow you to initiate the required operation, do something else and then wait for a result of the operation. When properly designed, this approach allows you to get the most out of your processor - useless blocking is minimized.

2. ConnectEx() and AcceptEx() Socket 2 APIs

   AcceptEx() is useful overlapping function that allows you to accept connection on the background and get the size of the message sent via ConnectEx() API in one step. We use these functions to optimize our code so we can allocate memory while we are still waiting for slow connection, for instance. Again, we can minimize useless blocking and waiting in our code while letting the processor do more job for us at the same time.

3. I/O Completion Ports

   I/O Completion Ports is an efficient mechanism that allows you to process multiple concurrent connections in one or multiple worker threads. It minimizes context switches needed to perform all operation. This approach is best scalable on Windows platform and therefore prefered, even though it is a bit more complicated.

4. Worker Threadpool

   To process messages as quickly as possible on multiprocessor systems, BBQ uses multiple pre-created worker threads. All these threads are ready and waiting to process messages immediately.

Extreme Cases

As you can see, useless blocking is minimized with a strong help of operating system and its provided asynchronous functions. That system provides a lot of buffers for socket requests, so we can postpone these requests in a queue and process them once we gain needed resources. However, these buffers aren’t unlimited, so in the extreme scenario where all resources are exhausted, we can do only one of two things: 1/ refuse the connection from server to client and fail to deliver the message or 2/ wait until some resources are freed - and it can be infinite waiting, or it will finally lead to a fail again. So, in the end no approach can finally solve the problem of limited resources and unlimited requests, however we tried to exploit system buffers as much as possible, so we pushed the moment of inevitable fail further away.

Design

BBQ server consits of a few parts:

1. Listening Thread

   listens for connections and passes them to processing thread

2. Processing Thread

   initiates receiving of message and passes it to one of available worker threads

3. Worker Threads

   retrieves messages and calls their proper message handler

You may ask, why is there the processing thread? Wouldn’t it be better to have everything in worker threads? The processing thread has its purpose: since most of the code is executed in worker threads (some blocking can happen in message handlers) and this happens asynchronously, it is possible to process another pending connections in a shorter time. Simply said, this concept improves overall server scalability again.

BBQ client is very simple and consists of just a single-threaded sender, however there are two things worth to mention:

1. Reconnection parameters

   In case the client is unable to deliver message in expected time, it lowers the reconnection attempts count and enlarges sleep time between reconnections. Otherwise on each successful delivery client adjusts these values contrarily until they reach original value. So in case server is flooded, client will stop sending messages after a few unsuccessful attempts, prolongs the wait time between them, and so enables the server to recover. Once this is done, client adjusts these parameters back. This is a kind of automatic load balancing feature.

2. Sending without waiting

   You can tell the clients that you don’t need them to wait for successful delivery. This can make the BBQ client work even faster, however you won't be notified when something goes wrong with message delivery.

Code sample

Now let’s take a detailed look at the BBQ server code bbq_server.cpp!BBQServerListeningThread(). We will start with the listening thread:

---
WSACreateEvent()
WSASocket()
memory allocation for new item
---
LOOP
	AcceptEx()                      - accepts connection
	---
	WSACreateEvent()
	WSASocket()
	memory allocation for next item
	---
	WSAWaitForMultipleEvents()      - now the connection is really accepted
	---
	add to queue for later processing
	replace old event, socket and queue item with the allocated ones
END LOOP

You can see that the code is designed a bit strangely. Why don’t we allocate events and sockets and item memory just when we need them? That’s because the allocation takes some time to finish. Listening thread pre-allocates needed resources that can be reused later, so we won’t be slowed by it when handling incoming messages. The allocation code is executed while we are actually accepting network connection, which is typically a slow operation that has to be finished anyhow. So, this is a typical usage of overlapping I/O operations, which makes BBQ more scalable.

Now let’s take a look at the processing thread bbq_server.cpp!BBQServerProcessingThread():

LOOP
	get queue item
	CreateIoCompletionPort(socket)
	memory allocation based on size we have received via AcceptEx()
	WSARecv()
	remove item from memory
END LOOP

The processing thread is even simpler than the listening one. It adds a socket to our I/O completion port, allocates memory for an incoming message, starts with receiving and lets one of available worker threads do the rest, so it can process another message immediately. We try to keep AcceptEx() loop in listening thread as small as possible.

And finally we have worker threads code here:

LOOP
	GetQueuedCompletionStatus()
	call message handler
	closesocket()
	release resources
END LOOP

When GetQueuedCompletionStatus() API succeeds, we have the message delivered. In the end, it’s not so difficult, huh? The operating system did the rest for us. The key to scalability is to have every piece of code ready to immediately do the job and wait for nothing.

Tests and measures

We have included performance/stress tests for server and client in our package (test/test-server and test/test-client projects and solutions). The client sends 4kB messages to the server and the server processes them in worker threads. The client works much faster than the server, so very soon the server becomes inevitably flooded.

On my Intel Core i5-2500 @ 3.30 GHz connected to Internet (so, some other traffic may occur there), with 4 server working threads, 4 client sending threads and 60 seconds for runtime, results looks like this:

SERVER

setting up ctrl+c handler...
initializing BBQ server...
creating BBQ server...
running BBQ server...
creating mutex...
waiting for termination (ctrl+c or 60000 miliseconds)...

runtime: 60062
server accepts: 64324 (1070.96 per sec)
server megabytes: 263.47 (4.39 per sec)
server receives: 64324 (1070.96 per sec)
max pending msgs: 1647
curr pending msgs: 0
Press any key to continue . . .

CLIENT

waiting for server mutex...
setting up ctrl+c handler...
creating working threads...
initializing BBQ client...
resuming threads...
waiting for termination...
send failed #1, thread: 8320, type: 6, err: 10055
send failed #2, thread: 14456, type: 6, err: 10055
send failed #3, thread: 12344, type: 6, err: 10055
send failed #4, thread: 5144, type: 6, err: 10055
send failed #5, thread: 8320, type: 6, err: 10055
send failed #6, thread: 14456, type: 6, err: 10055
send failed #7, thread: 12344, type: 6, err: 10055
send failed #8, thread: 5144, type: 6, err: 10055
send failed #9, thread: 8320, type: 6, err: 10055
send failed #10, thread: 14456, type: 6, err: 10055
send failed #11, thread: 12344, type: 6, err: 10055
send failed #12, thread: 5144, type: 6, err: 10055
send failed #13, thread: 14456, type: 6, err: 10055
send failed #14, thread: 12344, type: 6, err: 10055
send failed #15, thread: 5144, type: 6, err: 10055
send failed #16, thread: 8320, type: 6, err: 10055
send failed #17, thread: 12344, type: 6, err: 10055
send failed #18, thread: 5144, type: 6, err: 10055
send failed #19, thread: 8320, type: 6, err: 10055
send failed #20, thread: 14456, type: 6, err: 10055
send failed #21, thread: 12344, type: 6, err: 10055
send failed #22, thread: 5144, type: 6, err: 10055
send failed #23, thread: 8320, type: 6, err: 10055
send failed #24, thread: 14456, type: 6, err: 10055
send failed #25, thread: 12344, type: 6, err: 10055
send failed #26, thread: 5144, type: 6, err: 10055
send failed #27, thread: 8320, type: 6, err: 10055
send failed #28, thread: 14456, type: 6, err: 10055
send failed #29, thread: 14456, type: 6, err: 10055
send failed #30, thread: 8320, type: 6, err: 10055
send failed #31, thread: 12344, type: 6, err: 10055
send failed #32, thread: 5144, type: 6, err: 10055
send failed #33, thread: 14456, type: 6, err: 10055
send failed #34, thread: 8320, type: 6, err: 10055
send failed #35, thread: 12344, type: 6, err: 10055
send failed #36, thread: 5144, type: 6, err: 10055
send failed #37, thread: 14456, type: 6, err: 10055
send failed #38, thread: 8320, type: 6, err: 10055
send failed #39, thread: 12344, type: 6, err: 10055
send failed #40, thread: 5144, type: 6, err: 10055
send failed #41, thread: 14456, type: 6, err: 10055
send failed #42, thread: 8320, type: 6, err: 10055
send failed #43, thread: 12344, type: 6, err: 10055
send failed #44, thread: 5144, type: 6, err: 10055
send failed #45, thread: 14456, type: 6, err: 10055
send failed #46, thread: 8320, type: 6, err: 10055
send failed #47, thread: 12344, type: 6, err: 10055
send failed #48, thread: 5144, type: 6, err: 10055
send failed #49, thread: 14456, type: 6, err: 10055
send failed #50, thread: 8320, type: 6, err: 10055
send failed #51, thread: 12344, type: 6, err: 10055
send failed #52, thread: 5144, type: 6, err: 10055
send failed #53, thread: 14456, type: 6, err: 10055
send failed #54, thread: 8320, type: 6, err: 10055
send failed #55, thread: 12344, type: 6, err: 10055
send failed #56, thread: 5144, type: 6, err: 10055

runtime: 63969
clients sends: 64324 (1005.55 per sec)
clients megabytes: 263.47 (4.12 per sec)
clients fails: 56 (0.88 per sec)
connections fails: 184 (2.88 per sec)
queued msgs: 64324 (1005.55 per sec)
min sendtime: 0.00 secs
max sendtime: 5.03 secs
avg sendtime: 3.81 secs
Press any key to continue . . .

You can see that in 60 seconds the client was able to send 64.324 4kB long messages (total 263.47 MB), which is 4.14 MB/s. 56 times the delivery failed with error 10055 (WSAENOBUFS), meaning server is out of buffers. At the peak, server had to handle 1647 messages stored in a queue. We pushed sockets to its limits.

License and contacts

You can use and spread BBQ under BSD licence. If you have any questions, suggestions or improvements, you are free to contact me or any other Safetica guy.

Marek Strihavka aka Benny
marek.strihavka@safetica.com