| Index: branches/lucene-search-2.1/src/org/wikimedia/lsearch/search/SearcherCache.java |
| — | — | @@ -12,6 +12,7 @@ |
| 13 | 13 | import java.util.Comparator; |
| 14 | 14 | import java.util.HashSet; |
| 15 | 15 | import java.util.Hashtable; |
| | 16 | +import java.util.List; |
| 16 | 17 | import java.util.Set; |
| 17 | 18 | import java.util.Map.Entry; |
| 18 | 19 | |
| — | — | @@ -220,6 +221,10 @@ |
| 221 | 222 | protected static Set<String> initialWarmup = Collections.synchronizedSet(new HashSet<String>()); |
| 222 | 223 | |
| 223 | 224 | protected boolean initialDeploymentRunning = false; |
| | 225 | + |
| | 226 | + /** Number of threads to use for initial deployment */ |
| | 227 | + protected int initialDeploymentThreads = 1; |
| | 228 | + |
| 224 | 229 | /** |
| 225 | 230 | * If there is a cached local searcher of iid |
| 226 | 231 | * |
| — | — | @@ -452,19 +457,18 @@ |
| 453 | 458 | * Initialize all local searcher pools |
| 454 | 459 | */ |
| 455 | 460 | protected class InitialDeploymentThread extends Thread { |
| 456 | | - public void run(){ |
| 457 | | - try{ |
| 458 | | - initialDeploymentRunning = true; |
| 459 | | - IndexRegistry registry = IndexRegistry.getInstance(); |
| 460 | | - // get local search indexes, deploy sorted by name |
| 461 | | - ArrayList<IndexId> mys = new ArrayList<IndexId>(); |
| 462 | | - mys.addAll(GlobalConfiguration.getInstance().getMySearch()); |
| 463 | | - Collections.sort(mys,new Comparator<IndexId>(){ |
| 464 | | - public int compare(IndexId o1, IndexId o2) { |
| 465 | | - return o1.toString().compareTo(o2.toString()); |
| 466 | | - } |
| 467 | | - }); |
| 468 | | - for(IndexId iid : mys){ |
| | 461 | + IndexRegistry registry = null; |
| | 462 | + |
| | 463 | + protected class InitialDeployer extends Thread { |
| | 464 | + ArrayList<IndexId> iids = new ArrayList<IndexId>(); |
| | 465 | + |
| | 466 | + protected InitialDeployer(List<IndexId> iids){ |
| | 467 | + this.iids.addAll(iids); |
| | 468 | + } |
| | 469 | + |
| | 470 | + public void run(){ |
| | 471 | + log.info("Starting initial deployer for "+iids); |
| | 472 | + for(IndexId iid : iids){ |
| 469 | 473 | try { |
| 470 | 474 | // when searcher is linked into "search" path it's good, initialize it |
| 471 | 475 | if(!iid.isLogical() && registry.getCurrentSearch(iid) != null){ |
| — | — | @@ -480,6 +484,45 @@ |
| 481 | 485 | log.warn("I/O error warming index for "+iid+" : "+e.getMessage(),e); |
| 482 | 486 | } |
| 483 | 487 | } |
| | 488 | + } |
| | 489 | + } |
| | 490 | + public void run(){ |
| | 491 | + try{ |
| | 492 | + initialDeploymentRunning = true; |
| | 493 | + registry = IndexRegistry.getInstance(); |
| | 494 | + // get local search indexes, deploy sorted by name |
| | 495 | + ArrayList<IndexId> mys = new ArrayList<IndexId>(); |
| | 496 | + mys.addAll(GlobalConfiguration.getInstance().getMySearch()); |
| | 497 | + Collections.sort(mys,new Comparator<IndexId>(){ |
| | 498 | + public int compare(IndexId o1, IndexId o2) { |
| | 499 | + return o1.toString().compareTo(o2.toString()); |
| | 500 | + } |
| | 501 | + }); |
| | 502 | + int threadNum = initialDeploymentThreads; |
| | 503 | + ArrayList<InitialDeployer> threads = new ArrayList<InitialDeployer>(); |
| | 504 | + |
| | 505 | + // divide mys list into chunks and assign them to different worker threads |
| | 506 | + int inc = mys.size() / threadNum + 1; |
| | 507 | + int start = 0; |
| | 508 | + for(int i=0;i<threadNum;i++){ |
| | 509 | + threads.add(new InitialDeployer( |
| | 510 | + mys.subList(start, Math.min(start+inc, mys.size())))); |
| | 511 | + start += inc; |
| | 512 | + } |
| | 513 | + |
| | 514 | + // start all threads |
| | 515 | + for(InitialDeployer t : threads) |
| | 516 | + t.start(); |
| | 517 | + |
| | 518 | + // wait for all of the threads to finish |
| | 519 | + for(InitialDeployer t : threads) |
| | 520 | + try { |
| | 521 | + t.join(); |
| | 522 | + } catch (InterruptedException e) { |
| | 523 | + log.error("Thread "+t+" didn't finish properly", e); |
| | 524 | + } |
| | 525 | + |
| | 526 | + |
| 484 | 527 | } finally { |
| 485 | 528 | initialDeploymentRunning = false; |
| 486 | 529 | } |
| — | — | @@ -560,7 +603,9 @@ |
| 561 | 604 | } |
| 562 | 605 | |
| 563 | 606 | protected SearcherCache(boolean initialize){ |
| 564 | | - searchPoolSize = Configuration.open().getInt("SearcherPool","size",1); |
| | 607 | + Configuration config = Configuration.open(); |
| | 608 | + searchPoolSize = config.getInt("SearcherPool","size",1); |
| | 609 | + initialDeploymentThreads = config.getInt("SearcherPool", "initThreads",1); |
| 565 | 610 | if(initialize){ |
| 566 | 611 | initialDeploymentRunning = true; |
| 567 | 612 | new InitialDeploymentThread().start(); |