HN user
The driver does not allocate in the main message flow so GC does not factor in. Also the driver can run out of process and not be impacted by client processes.
Thanks for the feedback. I'll adjust my terminology to be more precise as it is the right thing to do. Doing things in the open is a great way to learn. I'm not shy of airing my laundry :-)
While one thread is in the action of rotation other producers return right away from the offer. The possibility of starvation occurs if the same thread each time it retried, that buffer has advanced to the next rotation. If adding 100 byte messages to a 128 MB buffer that is greater than a 1 in a million chance on each rotation. To have this continue then the probabilities have to be multiplied for the number of times you expect it to happen. So for ultimate starvation that gets crazy very very quickly ;-) Do you see it as more likely than that?
When I re-read the definitions I can see what you take from it. I think it comes down to what you consider as systemic progress. With respect to the preciseness of the definitions I have likely misinterpreted this. Is the system the algorithm itself or is it the system it lives in? I assumed the latter which may well be a mistake.
Each thread under the algorithm can perform their actions in a finite number of steps without ever blocking. This means the producers can continue to do other work. The consumer can continue to consume from other log buffers without being blocked and complete in a finite number of steps. If a producer is killed mid operation then no further progress can be made on that log buffer. If this is considered blocking then the algorithm is blocking and therefore not wait-free. It would need to be killed by another malicious thread for this to happen.
What is clear is that this algorithm gives the best latency profile of of all the measured messaging systems and the highest throughput. I now have the challenge of searching for a name that best describes its behaviour.
I've read the _The Art of Multiprocessor Programming_ and am happy to read it again. Maybe I have misinterpreted.
So if "killed mid-operation" must be supported then I don't see how many algorithms can be said to make progress. Take for example the Lamport SPSC queue[1], if the producer gets killed mid operation between steps 4 and 5 of the push operation. Then the data is in the queue but the consumer is blocked from ever seeing it with this line of reasoning. The Lamport SPSC queue is considered wait-free by the concurrency community I know. I base my reasoning on this. What if the producer takes a pause for a long time between steps 4 and 5 before continuing?
However if to be wait-free an algorithm must allow all other threads to continue using the data structure, not just continue making progress in other ways by being non-blocking and completing in a finite number of steps, then I stand corrected.
If wait-free must include coping with any thread being killed mid operation is there a term for being lock-free but also having all threads not block and complete in a finite number of steps for their interaction with the algorithm?
Writer two is not blocked, it is not waiting, it is not being obstructed. It is wait-free. It returns and can do whatever work it wishes.
However the data structure would be corrupt if writer one never completed. This is corruption and not the definition of wait-free.
You keep saying you are not an expert but keep pushing the point :-)
Try this. A process could have a rogue thread scribbling on memory and thus corrupting it. Such a process then that has this thread, by your definition, cannot be wait-free because any data structure can be corrupted and cause algorithms to fail. This is not how the concurrency community look at this.
*added below
Even a simple LOCK XADD instruction could have its results corrupted by this rogue thread after it writes before a consumer reads it.
If you look at the code the reader does not read the tail. It skips forward reading the length fields. It would never see the writer two message and it never blocks. It just thinks there are no new messages.
Please read the algorithm in code. No threads ever wait. :-)
I understand the point that we need to make software robust. No single algorithm can cope with every possible failure mode. That is why we need multiple instances run on multiple nodes to be resilient. This is independent from if this is a wait-free algorithm.
Thanks for the link.
If you look at the implementation no thread is ever stopped from completing. If a producing thread was killed mid operation by another malicious thread then the stream is broken. Other threads are never blocked. Other writers can add via a finite number of steps and the stream will back pressure when flow control kicks in. The consumer is never blocked - it simply never sees more messages. The algorithm meets the definition.
Unless I missed it that paper does not not state that the algorithm must cope with a malicious thread to be wait-free.
Try applying your thinking to this other respected MPSC wait-free queue.
http://www.1024cores.net/home/lock-free-algorithms/queues/in...
Can you point to a respected publication that states that is the definition of a wait-free algorithm, i.e. that other threads must help and it must cope with killing the thread externally?
By kill a thread I assume you mean use the OS to interrupt and terminate the thread from the outside while in such a simple algorithm and yet expect it to cope. Have you tried this on any java.util.concurrent classes and reported the "bugs" you have found? I'd be interested in the feedback you got. :-)
Can you expand on how you see the writer can "fail"? Also please relate this to other algorithms you see as wait-free but don't have your failure issue. The code is very simple and does not have external dependencies. It is a three step process of advance tail, copy in message, apply header. All within the implementation, all threads make progress and completes in a finite number of steps, and thus wait-free.
https://github.com/real-logic/Aeron/blob/master/aeron-client...
I asked the question if this is based on conjecture or experimental evidence. I'm happy to base my view on experimental evidence from my own research and testing.
I'm sorry but this analysis is completely wrong. I don't work for Azul but have used C4 and spent time studying it. C4 does not use transactional memory, it does however employ a very elegant lock-free algorithm which allows continuous concurrent compacting collections. BTW that is why it is called C4.
I can tell you from first hand experience that is does not have throughput issues. The truth is quite the opposite as it is the only JVM I've found scales well on large core count or large memory systems.
Do you have experimental evidence to back up your claims or are they just conjecture?
Having studied C4 it is important to note that to get the best out of it you need to allow it to avail of a large heap. If your mindset is to keep the heap small, as you would with many other collectors, then you will be restricting the C4 collecting algorithm from performing at its best. Don't be shy, give it a big heap to play with. The obvious consequence to this is that C4 is not ideal for constrained memory systems - but that is not the world the majority of our servers live in today.
Amendment: All the above applies to Zing (C4) on Intel x86_64.
x86 is a total store order memory model so any ASM MOV instruction that writes to a memory address will eventually be seen when the store buffer drains. In code we must ensure the write is not register allocated. This is achieved by the use of lazySet() as described in the article.
You are absolutely right that volatile is inadequate for ordering C/C++ concurrent algorithms, and memory barriers/fences are additionally required. I tried to focus on the hardware in this article.
I like to have only one writer thread for any data. This can have its own flyweight. Other threads can be readers each having their own read only flyweight implementation.
In finance you may be re-evaluating a whole portfolio of assets, or doing a value at risk (VAR) calculation across everything. More often you want low-latency access to the entire dataset without going to disk. For this the entire data must be memory resident and compact.
Interesting. It sounds like your issues are IO dominant since you do not mind the JVM startup cost from Hadoop for each query on each node. I'm more often looking at large data that is all memory resident which tends to drive the design this way. In finance queries need to have latencies way below sub-second which Hadoop cannot come close to satisfying. This is comparing batch to real-time analytics.
How would it work with object pooling if I wanted to query a large table of data? This is often needed in real big data applications.
Most of those frameworks you reference are much greater than 1 microsecond round trip which is the same ballpark as good network IO. For IPC this is much less. I operate in the financial low-latency space where this matters.
State of the art network hop is less than 3us so this is relevant. For on box IPC which can be as low as 100ns then very very relevant.
The byte[] buffer is deliberate because that is what will typically be passed to the network or IO device, so easier integration.
If you check the code you will see the buffer is only allocated once so the cost is not an issue.
A good posting showing the effect of setting thread affinity for version 1.0 of the Disruptor.
Single nodes can die in the system without issue. They often do! Since we use IP multicast the network failure is transparent as a replica takes up the primary role.
The one issue to be managed with this type of system is exceptions in the business logic thread. This can be handled via a number of prevention techniques. First, apply very strict validation on all input parameters. Second, take a test driven approach to development; at LMAX we have 10s of thousands of automated tests. Third, code so methods are either idempotent, or changes are only applied at the end when the business logic is complete using local variables. With this combination of approaches we have not seen a production outage due to business logic thread failure in over a year of live operation.
The business logic simply outputs state changes that other systems subscribe to. If the subscribing system misses the update our reliable message delivery will ensure the message gets to the subscribing system and processed. Our system guarantees message processing and not just delivery.
Queries against the business domain model are just input events. The business logic will serialise the requested part of the model and publish it out. This type of query can easily be handled by any number of replica nodes.
All messages in the system carry a sequence number so duplicate messages can be detected and ignored.
All transactions processed by the LMAX system are ACID. The application is modelled such that each input event represents one transaction. Where business transaction span multiple nodes, rather than handle the complexity of retaining ACID properties across multiple messages, our reliable messaging system ensures delivery to a remote system and queuing a message is included in the same transaction context as the business logic. This is true of services that use a high speed journal (event-sourced) and those that perform more traditional database operations. Long running transactions become steps in a state model.
The LMAX system is more about latency than throughput. When designing for very low-latency the result can be a system that achieves very great throughput if the appropriate techniques are employed. Single threaded applications are very suitable for low-latency because of the avoidance of lock contention and predictability they bring.
Some means of reliable delivery of messages, to and from this single thread, is necessary to make a useful application. These messages must be delivered in the event of system failures. To address this need the Disruptor is employed to pipeline and run in parallel, the replication, journalling and business logic for these messages. The whole system is asynchronous and non-blocking.
In our architecture we have multiple gateway nodes that handle border security and protocol translation to and from our internal IP multi-cast binary protocol for delivery to highly redundant nodes. Lots of external connections can be multiplexed down from the outside world this way.
The 6 million TPS refers to our matching engine business logic event processor. We have other business logic processors for things like risk management and account functions. These can all communicate via our guaranteed message delivery system that can survive node failures and restarts, even across data centres.
Modern financial exchanges can process over 100K TPS and have to respond with latency in the 100s of microseconds firewall to firewall, thus including the entire internal infrastructure. For those tracking the latest developments will see it is possible to have single digit microsecond network hops with IP multicast with user space network stacks and RDMA over 10GigE. Even a well tuned 1GigE stack can achieve sub 40us for a network hop. For reference single digit microseconds is in the same space as a context switch on a lock with the kernel arbitrating. Most financial exchanges rely on having data on multiple nodes before the transaction is secure. A number of these nodes can be asynchronously journalling the data down to disk. At LMAX we tend to have data in 3 or more nodes at any given time.
In my experience of profiling many business applications the vast majority of the time is either spent in protocol translation such as XML or JSON to business objects, or within the JDBC driver doing buffer copying and waiting on the database to respond, when the application domain is well modelled.
Often applications are not well modelled for their domain. This can result in algorithms that, rather than be O(1) for most transactions, have horrible scale up characteristics because of inappropriate collections representing relationships. If you have the luxury of developing an in-memory application requiring high performance it quickly becomes apparent the cost of a CPU cache miss is the biggest limitation to latency and throughput. For this one needs to employ data structures that exhibit good mechanical sympathy for CPU and memory subsystem. At LMAX we have replaced most of the JDK collections with our own that are cache friendly and garbage free.
So far we have had no issue processing all transactions for a given purpose on a single thread, or holding all the live state in memory for a single node. If we ever cannot process all the transactions necessary on a single thread then we simply shard the model across threads/execution contexts. We only hold the live mutating data in-memory and archive out to database completed transactions as they are then read only.
Martin (LMAX CTO)