Designing Software Around Data Grid and Compute Grid Programming Model

OpenCog discussion of impact of thread-safety and distributed processing on performance (and code structure):

On Sunday, July 6, 2014 5:51:36 PM UTC-4:30, linas wrote:

On 6 July 2014 11:51, Ramin Barati <rek...@gmail.com> wrote:

Anyway, a few years ago, the AtomSpace used the proxy pattern and it was a performance disaster. I removed it, which was a lot of hard, painful work, because it should never have been added in the first place.  Atomspace addnode/link operations got 3x faster, getting and setting truth values got 30x faster, getting outgoing sets got 30x faster.  You can read about it in opencog/benchmark/diary.txt

I can't even imagine what use has the Atomspace for the proxy pattern.

Ostensibly thread-safety, and distributed processing. Thread-safety, because everything was an "atom space request" that went through a single serialized choke point.  Distributed processing, because you could now insert a zeromq into that choke point, and run it on the network.  The zmq stuff was even prototyped, When measured, it did a few hundred atoms per second, so work stopped.

To me, it was an example of someone getting an idea, but failing to think it through before starting to code.  And once the code was written, it became almost impossible to admit that it was a failure.  Because that involves egos and emotions and feelings.

--linas

I think designing most parts of the software around data grid and compute grid constructs would allow:

1. intuitive coding. i.e. no special constructs or API, just closures and local data, and closures are still cheap even when lots of iterations and no thread switching. e.g.
   Range(0..100000).foreach { calculateHeavily(it); }.collect( sum )
   
   a nice side effect of this is, a new member or hire can join a project and ideally, not having to learn the intricacies of messaging & multithreading plumbing, there's already too much logic to learn anyway without adding those glue.
   
2. practical multithreading. I'm tempted to say painless multithreading :) multithreading becomes configuration, not a baked in logic that's "afraid" to be changed (in the sense that, while make a singlethreaded code to multithreaded takes time, I think it takes even more time to make it work right + no race conditions + avoid negative scaling.. then when all else fails you return it back to the original code, while other developments are done concurrently)
   
3. messaging. even the simple code above implies messaging, i.e. logically it is sending "int i" into the closure. at runtime it can be singlethread, multithread, or multinode. if calculateHeavily is in nanoseconds then it's pointless to use multithread. but if calculateHeavily takes more than 1 second multinode is probably good.
   
4. data affinity. the data passed to the closure doesn't have to be "data"/"content", it can be a key, which the closure can then load locally and process and aggregate.
   findAtomsWhichAre(Person).foreach( (personId) -> { person = get(personId); calculateHeavily(person); }.collect( stats )
   
   I haven't seen the ZeroMQ implementation of AtomSpace, but I'm suspecting it is (was?) a chatty protocol, and would've been different if data affinity was considered. I only ever used AMQP, but I think both ZeroMQ and AMQP are great to implement messaging protocol instead of scaling, unless explicitly designed as such like how Apache Storm uses ZeroMQ.
5. caching... consistently & deterministically. one way to reduce chattiness is to cache data you already have, that can be reused on next invocations (which hopefully many, otherwise too many cache misses will be worse). the problem is cache becomes stale. the solution is to use the data grid model, meaning the cache is consistent at all times. While this means there is a bit more coupling between cache and database, the programming model stays the same (as point #1) and no need for stale cache checks.
   
6. caching... transactionally. manually implementing atomicity is hard in using messaging/RPC/remote proxy, or even basic multithreading. a data grid framework allows transactional caching, so
   ids.foreach { a = get(id) -> process(a) -> put(id, a) }
   would not step on some other operation.
   
7. performance testing for several configurations. if there are performance unit tests for a project, these can be in multiple configs: 1 thread; n threads; 2 nodes × n threads.
   this ideally achieves instant gratification. if a performance unit test has negative scaling, you can notice it earlier. and if it does approach linear scaling, congrats & have a beer :D
   
8. bulk read & write. related to #5, if there are lots of scattered writes to database, a cache would improve this using write-through, while maintaining transactional behavior. instead of 100 writes of 1 document each, the cache can bulk-write 1 database request of 100 documents. you may let the framework do it or may code bulk write manually in certain cases, there's the choice.
   
9. bulk messaging. related to #3 and #4. a straightforward messaging protocol may divide 100 operations into send 50 messages to node1 and 50 messages to node2, which may be significant overhead. a compute grid can divide 100 operations into 2 messages: 1 message of 50 operations to node1 and 1 message of 50 operations to node2.
   
10. avoid premature optimization, while allowing both parallelization and optimization. related to point #7, since you know you can change config at any time, if you notice negative scaling you can simply set the config for that particular cache or compute to single-thread or single-node. Implementing #2 + #3 manually sometimes actually hinders optimization.

While my reasons above pertain to performance, I'm not suggesting considering performance while coding, but I do suggest considering a programming model which allow you to implement performance tweaks unobtrusively and even experimentally while retaining sanity. (that last part seems to be important at times :)

For example, using GridGain one uses closures or Java8 lambdas or annotated methods to perform compute, so things aren't that much different. To access data one uses get/put/remove/queries abstraction and you usually abstract these anyway, but now you have the option to make these read-through and write-through caches instead of direct to database. Data may be served from a remote/local database, a remote grid node, or local memory, this is abstracted. The choice is there, but it doesn't have to clutter code, and can be either implemented separately a in different class (Strategy pattern) or configured declaratively.

For Java it's easy to use GridGain, I believe modern languages like Python or Scheme also have a way to achieve similar programming model. Although if for C++ then I can't probably say much.

Personally I'd love for a project to evolve (instead of rewrite), i.e. refactoring different parts over versions while retaining general logic and other parts. This way not only retains code (and source history), but more importantly team knowledge sharing. In a rewrite it's hard not to repeat the same mistake, not to mention the second-system effect.

In my experience with Bippo eCommerce, our last complete rewrite was 2011 when we switched from PHP to Java. From that to the present we did several major architectural changes, as well as frameworks: JSF to quasi-JavaScript to Wicket, Java EE to OSGi to Spring, Java6 to Java7 to Java8, LDAP to MongoDB, MySQL to MongoDB to MongoDB + PostgreSQL, and so on... sure we had our share of mistakes but the valuable part is we never rewrite the entire codebase at once, we deprecated and removed parts of codebase as we go along. And the team retains collective knowledge of the process, i.e. dependency between one library and another, and when we change one architecture the other one breaks. I find that very beneficial.

Performance Unit Testing to Reduce Friction in Software Evolution

Dr. Linas Vepstas' concern regarding premature distribution architecture in OpenCog AtomSpace that hinders performance optimization:

Anyway, a few years ago, the AtomSpace used the proxy pattern and it was a performance disaster. I removed it, which was a lot of hard, painful work, because it should never have been added in the first place.  Atomspace addnode/link operations got 3x faster, getting and setting truth values got 30x faster, getting outgoing sets got 30x faster.  You can read about it in opencog/benchmark/diary.txt

--linas

With regard to performance, I think it's probably useful to have a performance-unit-test for each project. Just like unit tests but for timing iterations of operations with known data, instead of correctness.

Doesn't have to be lots of data, just a few curated set of input data that's somewhat representative.

This way development (and development branches) will have a history of simple performance metrics, on a regular basis.

The intention is not to encourage premature optimization, but to give early warning when a code restructure/refactoring might impact performance negatively.

With that out of the way, the developer is free-er to explore software design options, test feature branches, so that internal structures & implementation stay "sane".

(it would imply that the tests themselves would have maintenance cost and this would be alleviated by having a relatively-stable public API that the tests can rely on to.)

Distributing and Parallelizing Probabilistic Logic Networks Reasoning

During discussion about making Probabilistic Logic Networks (PLN) allow parallel reasoning using distributed AtomSpace, Dr. Ben Goertzel noted:

As for parallelism, I believe that logic chaining can straightforwardly be made parallel, though this involves changes to the algorithms and their behavior as heuristics.   For example, suppose one is backward chaining and wishes to apply deduction to obtain A --> C.   One can then evaluate multiple B potentially serving the role here, e.g.

A --> B1, B1 --> C  |-  A -->C

A --> B2, B2 --> C  |-  A -->C

...

Potentially, each Bi could be explored in parallel, right?    Also, in exploring each of these, the two terms could be backward chained on in paralle, so that e.g.

A --> B1

and

B1 --> C

could be explored in parallel...

In this way the degree of parallelism exploited by the backward chainer would expand exponentially during the course of a single chaining exploration, until reaching the natural limit imposed by the infrastructure.

This will yield behavior that is conceptually similar, though not identical, to serial backward chaining.

I haven't learned much about PLN yet, but I hope it can be made to work by distributing computation across AtomSpace nodes.

Other than strictly parallel, each path can be assigned a priority or heuristic, the tasks become a distributed priority queue. So a task finishing earlier can insert more tasks into the prority queue, and these new tasks don't have to be at the very end, but can be in the middle etc.

If we'd like to explore 1000 paths, and each of those generates another 1000 paths, ideally we'd have 1 million nodes. The first phase will be executed by 1000 nodes in parallel, the next 1 million paths will be executed by 1 million nodes in parallel, and we have a result in 2 ms. :) In practice we probably only can afford a few nodes, and limited time, can PLN use heuristic discovery?

For example, if the AI participates in "Are you smarter than a 5th grader", the discovery paths would be different than "calculate the best company strategy, I'll give you two weeks and detailed report". In a quiz, the AI would need to come up with a vague answer quickly, then refine the answer progressively until time runs out. i.e. when requested 10 outputs, the quiz one will try to get 10 answers as soon as possible even if many of them are incorrect; and the business one will try to get 1 answer correct, even if it means the other 9 is left unanswered.

Does PLN do this? If so, the distributed AtomSpace architecture would evolve hand-in-hand with (distributed) PLN. An app or modules shouldn't be required to be distributed to use AtomSpace, however a module (like PLN) that's aware that AtomSpace is both a distributed data grid and a distributed compute grid, can take advantage of this architecture and make its operations much faster/scalable. It's akin to difference between rendering 3D scenes by CPU vs. using OpenGL-accelerated graphics. However, a computer usually have only 1 or 2 graphics card and fixed, where an AtomSpace cluster can have dynamic number of nodes and you can throw more at it at any time. i.e. for expensive computation you can launch 100 EC2 instances for several hours then turn it off when done.

Adding Distributed Indexes to Hypergraph Database for Horizontal Scaling of Semantic Reasoning

While discussing distributed AtomSpace architecture in OpenCog group, Dr. Linas Vepstas noted:

Reference resolution, reasoning and induction might be fairly local as well:  when reading and trying to understand a wikipedia article, it seems as if its related to a million different things.   A single CPU with 16GB RAM can hold 100 million atoms in RAM, requiring no disk or network access.   

The only reason for a database then becomes as a place to store, over long periods of time, the results of the computation.  Its quite possible that fast performance of the database won't actually be important.  Which would mean that the actual database architecture might not be very important.  Maybe.

Based on the experiments, while processing (i.e. reasoning) 200,000 "atoms" in 3 seconds on a single host isn't too bad, searching for a few atoms out of 200,000 (or even 1 billion) on single host should take very fast (i.e. ~ 1 ms or less).

So I guess these are two distinct tasks. Searching would use (distributed) indexing, while processing/reasoning can be done by MindAgents combining data-to-compute and compute-to-data, with consideration to data affinity.

For processing which requires non-local data that Dr. Vepstas concerned, when using compute+data grid such as GridGain, a compute grid is automatically a cache, so all required non-local data are automatically cached. Which may or may not be sufficient, depending on the algorithm.

For searches, it seems we need to create separate indexes for each purpose, each index is sharded/partitioned appropriately to distribute compute load. Which means AtomSpace data grid is will have redundancy in many ways. The AtomSpace can probably be "split" into 3 parts:

1. the hypergraph part (can be stored in HyperGraphDB or Neo4j)
2. the eager index parts, always generated for the entire hypergraph, required for searches (can be stored in Cassandra or Solr or ElasticSearch)
3. the lazy index parts, the entries are calculated on demand then stored for later usage (can be stored in Cassandra or Solr or ElasticSearch)

The hypergraph would be good when you already know the handles, and for traversing. But when the task is "which handles A are B of the handles C assuming D is E?" an index is needed to answer this (particular task) quickly. Hopefully ~1 ms for each grid node, so 100 nodes working in parallel, will generate 100 set of answers in the ~1 ms.

Today, a 16 GB RAM node with 2 TB SATA storage is probably typical config (SSD will also work, but just for the sake of thought experiment a spinning disk more performance concerns). The node holds a partition of the distributed AtomSpace, and is expected to answer any search (i.e. give me handles of atoms in your node where it matches criteria X, Y, Z) within 1ms, and can do processing over a select nodes (i.e. for handles [A, B, C, ... N] perform this closure) within 1 second.

To achieve these goals:

1. For quick searches for that partition, all atom data needs to be indexed in multiple ways, an index for each purpose
2. For quick updates to the index (triggered by updates to data), the index and data are colocated in the same host to avoid network IO, although can be in different stores (i.e. data in HyperGraphDB and index in Cassandra). The partitioning/sharding need to accomodate this. So for 2 TB storage, we can put perhaps 100 GB data and 1 TB of indexes.
3. For quick lookup and updates of subset of data, the RAM is used as read-through & write-through cache by the data grid.
4. For non-local search/update/lookup/processing, it uses the data grid to do so, and caches results locally in RAM, that can overflow to disk. We still have 900 GB of space left, so we can use it for this purpose.
5. For quick processing of subset of data, local lookups are performed (which should take near-constant time, even with drives) and much faster if requested data is already in cache. Processing is then done using CPU or GPGPU (via OpenCL, e.g. Encog neural network library uses OpenCL to accelerate calculations). Results are then sent back via network.

For question-answering, given the label (e.g. Ibnu Sina) and possible concept types (Person), and optionally discussion contexts (Islam, religion, social, medicine), find the ConceptNode's which has that label, that type, and the confidence value for each contexts. And I want it done in 1 ms. :D

YAGO has 15,372,313 labels (1.1 GB dataset) for 10+ million entities. The entire YAGO is 22 GB. Assuming the entities with labels are stored in AtomSpace, selecting the matching labels without index would take ~150 seconds on a single host and ~50 seconds on 3 nodes (extrapolating my previous results). With indexes this should be 1ms.

First index would give the concepts given a label and types, with structure like :

label -> type -> [concept, concept, concept, ...]

            type -> [concept, concept, concept, ...]

            type -> [concept, concept, concept, ...]

Second index would give the confidence, given a concept and contexts, with sample data like :

Ibnu_Sina1 -> { Islam: 0.7, medicine: 0.9, social: 0.3, ... }

Ibnu_Sina2 -> { Islam: 0.1, medicine: 0.3, social: 0.9, ... }

Indexes change constantly, for each atom change multiple indexes must be updated, and index updates would take more resources than updating the atoms themselves, so index updates are asynchronous and eventually consistent. (I guess this also happens on humans, when humans learn new information, they don't immediately "understand" it. I mean, we now know a new fact, but it takes time [or even sleep] to make sense or implications/correlations of that new fact.)

We should agree on a set of a priori indexes. (As new concepts are learned and OpenCog gets queries that take a long time processing too many atoms, the AI may learn to make new indexes or tune existing ones... although this is probably too meta and distant future. :D )

Experimental Performance Test using GridGain for Distributed Natural Language Processing

I did an experimental performance test using GridGain to simulate AtomSpace processing. This is related to discussion in OpenCog group about AtomSpace architecture.

Disclaimer: This is not a benchmark, please don't treat it as such!

First I loaded up 212,351 YAGO labels (from MongoDB, but the actual backend doesn't matter here) for resources starting with letter M :

13:13:43.178 [main] INFO  i.a.i.e.l.l.yago.YagoLabelCacheStore - Loading 212351 labels...

13:13:45.595 [main] DEBUG i.a.i.e.l.l.yago.YagoLabelCacheStore -   [23%] 50000 labels loaded, 162351 more to go...

13:13:47.571 [main] DEBUG i.a.i.e.l.l.yago.YagoLabelCacheStore -   [47%] 100000 labels loaded, 112351 more to go...

13:13:49.139 [main] DEBUG i.a.i.e.l.l.yago.YagoLabelCacheStore -   [70%] 150000 labels loaded, 62351 more to go...

13:13:50.608 [main] DEBUG i.a.i.e.l.l.yago.YagoLabelCacheStore -   [94%] 200000 labels loaded, 12351 more to go...

13:13:50.914 [main] INFO  i.a.i.e.l.l.yago.YagoLabelCacheStore - Loaded 212351 labels...

13:13:50.917 [main] INFO  id.ac.itb.ee.lskk.lumen.yago.Worker - For yagoLabel, I have 100000 primary out of 100000 entries + 112351 swap

To make it somewhat more realistic, grid data for a node is capped at 100,000 entries. The configuration is partitioned, so for all 3 nodes then the entire dataset should be held entirely in memory. Then I started two more nodes, and the latest node does a search which resource ID has the label "Muhammad". So it's basically a reverse hashmap lookup, that can be perfectly be done using an index. But I'm treating the entries as atoms, just for the sake of doing distributed-parallel computation on them.

Collection<Set<String>> founds = labelCache.queries().createScanQuery(null).execute(new GridReducer<Entry<String, String>, Set<String>>() {

Set<String> ids = new HashSet<>();

@Override

public boolean collect(Entry<String, String> e) {

if (e.getValue().equalsIgnoreCase(upLabel)) {

ids.add( e.getKey() );

}

return true;

}

@Override

public Set<String> reduce() {

return ids;

}

}).get();

The results, using my workstation i5-3570K @ 4x 3.40GHz, 3 nodes at 1 GB heap each:

[13:29:40] GridGain node started OK (id=03a07172)

[13:29:40] Topology snapshot [ver=5, nodes=3, CPUs=4, heap=3.0GB]

13:29:40.043 [main] INFO  i.a.i.e.l.l.yago.YagoLabelLookupCli2 - Finding resource for label 'Muhammad'...

13:29:43.131 [main] INFO  i.a.i.e.l.l.yago.YagoLabelLookupCli2 - Found for Muhammad: [[Muhammad_Khalil_al-Hukaymah, Muhammad_S._Eissa, Muhammad_Musa, Muhammad_Okil_Musalman, Muhammad_Loutfi_Goumah, Muhammad_Sadiq, Muhammad_Salih, Muhammad_Ismail_Agha, Muhammad_Yusuf_Hashmi, Mustafah_Muhammad, Muhammad_Mahbubur_Rahman, Muhammad_Ahmad_Said_Khan_Chhatari, Muhammad_Jamiruddin_Sarkar, Muhammad_Ibrahim_Joyo, Muhammad_bin_Tughluq, Muhammad_Sohail_Anwar_Choudhry, Muhammad_Tariq_Tarar], [Muhammad_Salman, Muhammad_Jailani_Abu_Talib, Muhammad_Qutb], [Muhammad_Ibrahim_Kamel, Muhammad_Amin_Khan_Turani, Muhammad_Ali_Pate, Muhammad_Rafi_Usmani, Muhammad_Faisal, Muhammad, Muhammad_Ilham, Muhammad_Kurd_Ali, Muhammad_Umar, Muhammad_Shahidullah, Muhammad_Anwar_Khan, Muhammad_Saifullah, Muhammad_Saqlain]]

[13:29:43] GridGain node stopped OK [uptime=00:00:03:865]

Searched 212,351 entries in 3088 ms, using 3 nodes × 4 threads = 12 total threads on single host. So the rate is ~68766 entries/second.

To be fair, GridGain is giving performance hints: (so for serious benchmark, these should be tuned)

[13:29:40]   ^-- Decrease number of backups (set 'keyBackups' to 0)

[13:29:40]   ^-- Disable fully synchronous writes (set 'writeSynchronizationMode' to PRIMARY_SYNC or FULL_ASYNC)

[13:29:40]   ^-- Enable write-behind to persistent store (set 'writeBehindEnabled' to true)

[13:29:40]   ^-- Disable query index (set 'queryIndexEnabled' to false)

[13:29:40]   ^-- Disable peer class loading (set 'peerClassLoadingEnabled' to false)

[13:29:40]   ^-- Disable grid events (remove 'includeEventTypes' from configuration)

Of course, 12 threads running on a single host isn't optimal, and there's no network saturation effects since all nodes are on the same host.

From how GridGain works, the performance should be (much?) better when there are 3 actual nodes/processors to work on. The key thing is that the calculation (map/reduce) is done on each node, so the "Mind Agent" (node 3) here only does roughly ~33% of the job, the other 2 "AtomSpace" nodes aren't just serving data, they're also processing data they already have, no need to move these bits around the network.

Since the closure is code (Java code), it's possible to use OpenCL/GPU for certain tasks, which should increase performance for math-intensive processing.

Fault tolerance also works very well, so you can kill and rearrange nodes at will, the grid will stay there as long as at least 1 node is up.