Multi-threaded indexing in DSE Search 3.0.1
Until version 3.0, DSE Search provided the same index concurrency model as plain Apache Solr: all index update requests are handled by a single thread, which synchronously writes into Cassandra and then into the Lucene index via the integrated Cassandra secondary index mechanism. This is all good: the client is on duty to provide the desired concurrency level by issuing multiple concurrent requests, and DSE Search follows by processing each request on its own thread.
But, there are cases when the model above doesn’t provide enough performance, more specifically:
- Reindexing: DSE Search data can be reindexed “in place”, without forcing the client to resubmit data; this happens by going through Cassandra SSTables and sequentially indexing all rows.
- Repairing: Cassandra data can be repaired via the nodetool command, which triggers the reindexing of actually repaired rows by sequentially going through them.
- Bulk loading: many users implement home-made bulk loading solutions, which usually provide limited concurrency.
Multi-threaded indexing in DSE Search 3.0.1 comes to help in such a cases, also providing a general, improved, concurrent indexing model.
The new indexing model decouples writes to Cassandra from writes to the Lucene index, making the latter asynchronous. Conceptually speaking, it simply works as follows: when a document is inserted, its data is written into Cassandra and the indexing request is queued-up to be asynchronously processed by a pool of worker threads; at given intervals, when Memtable data is flushed from inside Cassandra, or a commit happens from inside Solr, the indexing queue is synchronously flushed, so that all in-flight indexing requests become visible and all committed data up to that point can be queried.
As simple as it sounds, when implementing asynchronous work models a few important problems have to be addressed:
- Concurrent execution of the logically same work unit: i.e., what if two asynchronous threads try to index the same document?
- Flow control between work producers and consumers: i.e., what if indexing requests are submitted faster than they are processed?
- Visibility and management: how can I know about my producers and consumers work? How can I tune it?
Let’s see how DSE Search 3.0.1 solves all of them.
First, we implement per-document thread affinity to avoid concurrent indexing of the same document: we hash the document identifier and assign the document to an in-memory queue serving one and only one indexing thread; by doing so, we partition the work between different threads, and make sure to always process indexing requests for the same document from the same thread.
Flow control is implemented via automatic back-pressure: at Cassandra flush or Solr commit time, we compute some heuristics representing the current load of the indexing system, based on index processing time and indexing queue depth, and if a configurable threshold is met, we pause producers (that is, incoming indexing requests) until all accumulated in-flight indexing requests are processed: please note this is different from a normal “flush situation”, when newly arrived indexing requests are allowed to queue-up and in-flight requests are flushed only up to the current point in time.
Visibility and management are implemented via configuration switches and JMX mbeans. First, you can configure the maximum number of indexing threads per core with the following dse.yaml setting: max_solr_concurrency_per_core; the default value is computed based on the available number of CPUs (which is usually a good value to stick on), but you can configure it to best suit your indexing volume, as well as set to 1 to get back to the old synchronous behavior. Then, for each Solr core, you have several configuration knobs and monitoring gauges available via the IndexPool-core_name mbean (where core_name is an actual Solr core name) in the com.datastax.bdp domain, with the more relevant ones being:
- BackPressureThreshold: the (average) max number of queued documents that will trigger the back-pressure mechanism discussed above; this is in other words a way to control and limit memory consumption of the whole indexing system.
- MaxConcurrency: the maximum number of indexing threads, between the previously mentioned max_solr_concurrency_per_core and 1; this way you can dynamically adjust your concurrency level, and even get back to the old synchronous model, without restarting.
- QueueDepth: current depth of all queues.
- TaskProcessingTime: time it took to process the latest indexing request, per queue/thread.
- ProcessedTasks: total number of processed tasks, per queue/thread.
Finally, some performance considerations: in our tests we’ve got a 30%-40% performance increase in sequential indexing use cases like repair and reindex, but it’s all about your use case in the end. With the visibility and configurability discussed before you have (hopefully) all the right tools for a correct analysis and effective tuning, but if in need of any help, do not hesitate to reach us via our public forums.