Multithreaded TGNI Concept Loader

Sometime back, I mentioned that I tried to load up our taxonomy (with about 1 million medical concepts), into TGNI's Lucene and Neo4J datastores, and the process took 3 weeks to complete (on my 2 CPU desktop at work, as a single threaded process). I've been meaning to see if I could speed it up, but the data was adequate for most of the experiments I was doing, so I did not have enough incentive. Until about 4 weeks ago, when I discovered that I had inadverdently pulled in retired and experimental concepts and that they were interfering with the quality of my output.

My initial plan was to convert the loading process into a Map-Reduce job with Hadoop, but I would have to server-ize Lucene and Neo4j (ie, using SOLR and Neo4j's REST API), and the prospect of having to start up 3 servers to test the application seemed a bit daunting, so I scrapped that idea in favor of just multi-threading the loading application. Although, in retrospect, that would have worked equally well (in terms of effort involved to implement) and would have been more scalable (in terms of the hardware requirements - its far easier to get a bank of low-powered servers than it is to get a single high-powered server).

In this post, I describe the somewhat convoluted process that led to a successful multi-threaded loader implementation, hoping that somewhere in this, there are lessons for people (like myself and possibly a vast majority of Java programmers) to whom writing non-trivial multithreaded apps is like buying a car, ie, something you have to do only once every say 5-7 years.

To provide some context, here is what the flow in my original (single threaded) loader looked like. The code would loop through a bunch of tables in an Oracle database and build concept objects out of it, then send the object to a node service, which consisted of a graph service and an index service. The concept would be added to the Neo4j graph database (and get a node ID in the process), then it would be sent to the index service, which would pass it through the UIMA/Lucene analyzer chain to create an entry (heavily augmented with attributes) in the Lucene index for each name (primary, qualified, synonyms) associated witht he concept.

My first implementation was to build a list of OIDs from the Oracle database, then spawn a fixed size thread pool using Java's ExecutorService. Each thread would then build a TConcept object, write to Neo4j, normalize the names and add them (as distinct entities) to the MySQL database. This would run through about 3,000 concepts before hanging. Thinking that perhaps it was something to do with the way I had integrated UIMA with Lucene analyzers, I broke them apart so the UIMA Analysis Engine (AE) would annotate each input name, then break them apart into (potentially) multiple strings, then feed them in, one by one, into the Lucene analyzer chain consisting of streaming Lucene only components (keyword attribute aware LowerCaseFilter, StopFilter and PorterStemFilter).

While I was doing this, I decided to switch out Lucene and use MySQL instead. I was pre-normalizing the names anyway, and I needed to match normalized versions of my input against normalized versions of the concept names. Using Lucene wasn't buying me anything - it was actually hurting because it would match partial strings, and I was having to write code to prevent that.

However, the pipeline would still hang at around the same point. I remembered that I had used Jetlang some time back, and decided to see if modeling it as a Jetlang actor would help. This version ran through about 70,000 concepts before it hung. While I was running this version, I noticed that the CPUs ran a lot cooler (using top and looking at the user CPU consumed) with the Jetlang version compared to my original multithreaded version.

At that point I realized that each of my threads in my original version was creating its own version of the UIMA AE, Lucene Analyzer and database Connection objects for each concept. Since Jetlang uses the Actor model, its threads were basically mini-servers that looped in a read-execute loop.

In an attempt to keep the code mostly intact (I was trying to reuse code as far as possible), I factored out these resources into pools using Commons-Pool and replaced the constructor (and destructor) calls with calls to pool.borrowObject() and pool.returnObject(). This helped, in the sense that I noticed less CPU utilization, but the job would just mysteriously block at around the same point, ie, no movement in the logs, top showing no activity except in one or two CPUs.

Digging deeper, I found that chemical names were being caught by my semantic hyphen transformation pattern (meant to expand hyphenated words into two word and single word tokens), and were generating thousands of synonyms for them.

At the same time, I realized that I could dispense with the pools altogether by modeling my threads as mini-servers (with a for(;;) loop breakable with a poison pill message) and giving each thread its own copy of an UIMA AE, Analyzer, Oracle and MySQL Connection objects. Neo4j allows only a single connection to the database, but is thread-safe, so I wrapped the connection in a singleton and gave each mini-server a reference to the singleton.

For chemical names, I put in an additional AE and changed the flow so if a string (or part of it) was already annotated, a downstream AE will not attempt to annotate it. However, just in case there were other wierd patterns lurking in the input, I wanted to be able to terminate the normalization process (and not process the concept) if it took "too long" to execute, so it did not hold up other concepts that could be processed.

With all these requirements, I ended up modeling the job in three levels - the manager which instantiates everything and creates a queue of input ids to process, a pool of worker threads which are mini-servers and which have their own instances of expensive resources, and normalization tasks, which are instantiated as callable futures from within the worker threads, and which timeout after a configurable amount of time (default 1s), and cause the UIMA CAS (an expensive resource that should be destroyed according to the UIMA docs) to be released and the AE rebuilt with a new CAS when that happens.

Here's the code (with the application specific stuff elided to keep it short, since it adds nothing to the discussion).

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440

// Source: src/main/java/com/mycompany/tgni/loader/ConceptLoadManager.java package com.mycompany.tgni.loader; import java.io.File; import java.io.Reader; import java.math.BigDecimal; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import opennlp.tools.util.Pair; import org.apache.commons.collections15.CollectionUtils; import org.apache.commons.lang.StringUtils; import org.apache.lucene.analysis.Analyzer; import org.apache.lucene.analysis.TokenStream; import org.apache.lucene.analysis.core.LowerCaseFilter; import org.apache.lucene.analysis.en.PorterStemFilter; import org.apache.lucene.analysis.standard.StandardTokenizer; import org.apache.lucene.util.Version; import org.apache.uima.analysis_engine.AnalysisEngine; import org.apache.uima.jcas.JCas; import org.neo4j.graphdb.GraphDatabaseService; import org.neo4j.graphdb.Node; import org.neo4j.graphdb.Transaction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; //import org.springframework.util.StopWatch; import com.mycompany.tgni.beans.TConcept; import com.mycompany.tgni.lucene.StopFilter; import com.mycompany.tgni.neo4j.GraphInstance; import com.mycompany.tgni.neo4j.JsonUtils; import com.mycompany.tgni.neo4j.NameNormalizer; import com.mycompany.tgni.uima.utils.UimaUtils; public class ConceptLoadManager { private final Logger logger = LoggerFactory.getLogger(getClass()); private static final int NUM_WORKERS = Math.round(1.4F * Runtime.getRuntime().availableProcessors()); private static final long TASK_TIMEOUT_MILLIS = 1000L; private static final CountDownLatch LATCH = new CountDownLatch(NUM_WORKERS); private static final BlockingQueue<Integer> QUEUE = new LinkedBlockingQueue<Integer>(); // oracle queries private static final String LIST_OIDS_SQL = "..."; private static final String GET_HEAD_SQL = "..."; private static final String GET_PNAMES_SQL = "..."; private static final String GET_SYNS_SQL = "..."; private static final String GET_STY_SQL = "..."; // mysql queries private static final String ADD_NAME_SQL = "insert into oid_name (" + "oid, name, pri) " + "values (?,?,?)"; private static final String ADD_NID_SQL = "insert into oid_nid (oid, nid) values (?, ?)"; public static void main(String[] args) throws Exception { // extract parameters from command line if (args.length != 5) { System.out.println("Usage: ConceptLoadManager " + "/path/to/graph/dir /path/to/mysql-properties " + "/path/to/stopwords/file /path/to/ae/descriptor " + "/path/to/oracle-properties"); System.exit(-1); } // Initialize manager ConceptLoadManager manager = new ConceptLoadManager(); final GraphInstance neo4jConn = new GraphInstance(args[0]); final String mysqlProps = args[1]; final Set<?> stopwords = StopFilter.makeStopSet( Version.LUCENE_40, new File(args[2])); final String aeDescriptor = args[3]; final String oraProps = args[4]; // seed input queue manager.seed(oraProps); // add poison pills for (int i = 0; i < NUM_WORKERS; i++) { try { QUEUE.put(-1); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } // set up worker threads ExecutorService workerPool = Executors.newFixedThreadPool(NUM_WORKERS); for (int i = 0; i < NUM_WORKERS; i++) { ConceptLoadWorker worker = new ConceptLoadManager().new ConceptLoadWorker( i, mysqlProps, stopwords, aeDescriptor, oraProps, neo4jConn); workerPool.execute(worker); } // wait for all tasks to process, then shutdown workerPool.shutdown(); try { LATCH.await(); } catch (InterruptedException e) { /* NOOP */ } neo4jConn.destroy(); workerPool.awaitTermination(1000L, TimeUnit.MILLISECONDS); } private void seed(String oraProps) { List<Integer> oids = new ArrayList<Integer>(); Connection conn = null; PreparedStatement ps = null; ResultSet rs = null; try { conn = DbConnectionUtils.getConnection(oraProps); ps = conn.prepareStatement(LIST_OIDS_SQL); rs = ps.executeQuery(); while (rs.next()) { QUEUE.put(rs.getInt(1)); } } catch (Exception e) { logger.warn("Can't generate OIDs to process", e); } finally { DbConnectionUtils.closeResultSet(rs); DbConnectionUtils.closePreparedStatement(ps); DbConnectionUtils.closeConnection(conn); } } /////////////// Worker Class /////////////////// private class ConceptLoadWorker implements Runnable { private int workerId; private AtomicInteger count; private int totalTasks; private Set<?> stopwords; private String mysqlProps; private String aeDescriptor; private String oraProps; private GraphInstance neo4jConn; private Connection mysqlConn; private PreparedStatement psAddNames, psAddNid; private Connection oraConn; private PreparedStatement psGetHead, psGetNames, psGetSyns, psGetSty; private AnalysisEngine ae; private JCas jcas; private Analyzer analyzer; public ConceptLoadWorker(int workerId, String mysqlProps, Set<?> stopwords, String aeDescriptor, String oraProps, GraphInstance neo4jConn) { this.workerId = workerId; this.count = new AtomicInteger(0); this.totalTasks = QUEUE.size(); this.mysqlProps = mysqlProps; this.stopwords = stopwords; this.aeDescriptor = aeDescriptor; this.oraProps = oraProps; this.neo4jConn = neo4jConn; } @Override public void run() { try { initWorker(); ExecutorService taskExec = Executors.newSingleThreadExecutor(); for (;;) { Integer oid = QUEUE.take(); if (oid < 0) { break; } int curr = count.incrementAndGet(); // load the concept by OID from oracle TConcept concept = null; try { concept = loadConcept(oid); } catch (SQLException e) { logger.warn("Exception retrieving concet (OID:" + oid + ")", e); continue; } // normalize names using UIMA/Lucene chains. This is // a slow process so we want to time this out if it // takes too long. In that case, the node/oid mapping // will not be written out into Neo4J. NameNormalizer normalizer = new NameNormalizer(ae, analyzer, jcas); NameNormalizerTask task = new NameNormalizerTask( concept, normalizer); Future<List<Pair<String,Boolean>>> futureResult = taskExec.submit(task); List<Pair<String,Boolean>> result = null; try { result = futureResult.get( TASK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); } catch (ExecutionException e) { logger.warn("Task (OID:" + oid + ") skipped", e); reinitWorker(); continue; } catch (TimeoutException e) { futureResult.cancel(true); logger.warn("Task (OID:" + oid + ") timed out", e); reinitWorker(); continue; } try { // add the OID-Name mappings to MySQL addNames(oid, result); // add the OID-NID mapping to Neo4j writeNodeConceptMapping(concept); } catch (Exception e) { logger.warn("Exception persisting concept (OID:" + oid + ")", e); continue; } // report on progress if (curr % 100 == 0) { logger.info("Worker " + workerId + " processed (" + curr + "/" + totalTasks + ") OIDs"); } } taskExec.shutdownNow(); } catch (InterruptedException e) { logger.error("Worker:" + workerId + " Interrupted", e); } catch (Exception e) { logger.error("Worker:" + workerId + " threw exception", e); } finally { destroyWorker(); LATCH.countDown(); } } private TConcept loadConcept(Integer oid) throws SQLException { TConcept concept = new TConcept(); // bunch of SQLs run against Oracle database to populate // the concept ... return concept; } private void addNames(Integer oid, List<Pair<String, Boolean>> names) throws SQLException { if (names == null) return; try { psAddNames.clearBatch(); for (Pair<String,Boolean> name : names) { if (StringUtils.length(StringUtils.trim(name.a)) > 255) { continue; } psAddNames.setInt(1, oid); psAddNames.setString(2, name.a); psAddNames.setString(3, name.b ? "T" : "F"); psAddNames.addBatch(); } psAddNames.executeBatch(); mysqlConn.commit(); } catch (SQLException e) { mysqlConn.rollback(); throw e; } } private void writeNodeConceptMapping(TConcept concept) throws Exception { logger.info("Writing concept (OID=" + concept.getOid() + ")"); GraphDatabaseService graphService = neo4jConn.getInstance(); Transaction tx = graphService.beginTx(); try { // update neo4j Node node = graphService.createNode(); concept.setNid(node.getId()); node.setProperty("oid", concept.getOid()); node.setProperty("pname", concept.getPname()); node.setProperty("qname", concept.getQname()); node.setProperty("synonyms", JsonUtils.listToString(concept.getSynonyms())); node.setProperty("stycodes", JsonUtils.mapToString(concept.getStycodes())); node.setProperty("stygrp", StringUtils.isEmpty( concept.getStygrp()) ? "UNKNOWN" : concept.getStygrp()); node.setProperty("mrank", concept.getMrank()); node.setProperty("arank", concept.getArank()); node.setProperty("tid", concept.getTid()); // update mysql psAddNid.setInt(1, concept.getOid()); psAddNid.setLong(2, concept.getNid()); psAddNid.executeUpdate(); mysqlConn.commit(); tx.success(); } catch (Exception e) { mysqlConn.rollback(); tx.failure(); logger.info("Exception writing mapping (OID=" + concept.getOid() + ")"); throw e; } finally { tx.finish(); } } private void initWorker() throws Exception { logger.info("Worker:" + workerId + " init"); // mysql this.mysqlConn = DbConnectionUtils.getConnection(mysqlProps); this.mysqlConn.setAutoCommit(false); this.psAddNames = mysqlConn.prepareStatement(ADD_NAME_SQL); this.psAddNid = mysqlConn.prepareStatement(ADD_NID_SQL); // oracle this.oraConn = DbConnectionUtils.getConnection(oraProps); this.psGetHead = oraConn.prepareStatement(GET_HEAD_SQL); this.psGetNames = oraConn.prepareStatement(GET_PNAMES_SQL); this.psGetSyns = oraConn.prepareStatement(GET_SYNS_SQL); this.psGetSty = oraConn.prepareStatement(GET_STY_SQL); // uima/lucene this.ae = UimaUtils.getAE(aeDescriptor, null); this.analyzer = getAnalyzer(stopwords); this.jcas = ae.newJCas(); } private void destroyWorker() { // mysql DbConnectionUtils.closePreparedStatement(psAddNames); DbConnectionUtils.closePreparedStatement(psAddNid); DbConnectionUtils.closeConnection(this.mysqlConn); // oracle DbConnectionUtils.closePreparedStatement(psGetHead); DbConnectionUtils.closePreparedStatement(psGetNames); DbConnectionUtils.closePreparedStatement(psGetSyns); DbConnectionUtils.closePreparedStatement(psGetSty); DbConnectionUtils.closeConnection(this.oraConn); // uima/lucene this.ae.destroy(); this.analyzer.close(); this.jcas.release(); this.jcas.reset(); } private void reinitWorker() throws Exception { this.ae.destroy(); this.analyzer.close(); this.jcas.release(); this.jcas.reset(); this.ae = UimaUtils.getAE(aeDescriptor, null); this.analyzer = getAnalyzer(stopwords); this.jcas = ae.newJCas(); } private Analyzer getAnalyzer(final Set<?> stopwords) { return new Analyzer() { @Override public TokenStream tokenStream(String fieldName, Reader reader) { TokenStream input = new StandardTokenizer(Version.LUCENE_40, reader); input = new LowerCaseFilter(Version.LUCENE_40, input); input = new StopFilter(Version.LUCENE_40, input, stopwords);; input = new PorterStemFilter(input); return input; } }; } } ///////////////// Task class //////////////// private class NameNormalizerTask implements Callable<List<Pair<String,Boolean>>> { private TConcept concept; private NameNormalizer normalizer; public NameNormalizerTask(TConcept concept, NameNormalizer normalizer) { this.concept = concept; this.normalizer = normalizer; } @Override public List<Pair<String,Boolean>> call() throws Exception { logger.info("Executing task (OID:" + concept.getOid() + ")"); Set<String> uniques = new HashSet<String>(); Set<String> normalizedUniques = new HashSet<String>(); List<Pair<String,Boolean>> results = new ArrayList<Pair<String,Boolean>>(); String pname = concept.getPname(); if (StringUtils.isNotEmpty(pname) && (! uniques.contains(pname))) { List<String> normalized = normalizer.normalize(pname); uniques.add(pname); normalizedUniques.addAll(normalized); } String qname = concept.getQname(); if (StringUtils.isNotEmpty(qname) && (! uniques.contains(qname))) { List<String> normalized = normalizer.normalize(qname); uniques.add(qname); normalizedUniques.addAll(normalized); } for (String normalizedUnique : normalizedUniques) { results.add(new Pair<String,Boolean>(normalizedUnique, true)); } Set<String> normalizedUniqueSyns = new HashSet<String>(); normalizedUniqueSyns.addAll(normalizedUniques); List<String> syns = concept.getSynonyms(); for (String syn : syns) { if (StringUtils.isNotEmpty(syn) && (! uniques.contains(syn))) { List<String> normalizedSyn = normalizer.normalize(syn); uniques.add(syn); normalizedUniqueSyns.addAll(normalizedSyn); } } Collection<String> normalizedSyns = CollectionUtils.subtract( normalizedUniques, normalizedUniqueSyns); for (String normalizedSyn : normalizedSyns) { results.add(new Pair<String,Boolean>(normalizedSyn, false)); } return results; } } }

Since the worker threads were doing a combination of IO (reading from the Oracle database and writing to MySQL and Neo4j) and CPU bound work (normalizing with the UIMA AE and Lucene Analyzers), I ran some timings on a small sample of 1000 concepts and found that it spent approximately 30% of its time doing IO. So based on the formula in Java Concurrency in Practice book:

1	num_threads = num_cpus * target_cpu_utilization * (1 + wait/compute)

I set the number of worker threads to 22 on my 16 CPU machine. During the run, I noticed that the load average was between 3-4 (which is quite low for a 16 CPU box) and the user CPU utilization percentages hovered in the 2-3% mark on most but 2-3 CPUs, which showed around 40-50% utilization. So there is probably still some room for increasing the number of worker threads. Here is a screenshot of top while the program is running.

With 22 threads, the job finished in a very acceptable time of about 1.5 hours, with 88 concepts timing out. I plan to look at those concepts to see if I can uncover patterns that would lead to the creation of some more pre-emptive AEs in the future.

Meanwhile, I hope I'll remember this stuff the next time I need to build one of these things :-). Its almost Christmas, so for those of you who celebrate it, heres wishing you a very Merry Christmas!