Merge branch 'branch-1.2.0' of https://github.com/tencent/angel into …

…branch-1.2.0
Angel-ML · Sep 13, 2017 · 3b36cb6 · 3b36cb6
2 parents 0c81658 + 0d984da
commit 3b36cb6
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 1 deletion.
diff --git a/angel-ps/core/src/main/java/com/tencent/angel/conf/AngelConf.java b/angel-ps/core/src/main/java/com/tencent/angel/conf/AngelConf.java
@@ -640,6 +640,11 @@ public AngelConf(Configuration conf) {
     + "matrixtransfer.server.user.sender";
   public static final boolean DEFAULT_ANGEL_MATRIXTRANSFER_SERVER_USER_SENDER = false;
 
+  public static final String ANGEL_MATRIX_OPLOG_MERGER_POOL_SIZE = ANGEL_PREFIX
+    + "matrix.oplog.merger.pool.size";
+
+  public static final int DEFAULT_ANGEL_MATRIX_OPLOG_MERGER_POOL_SIZE = Math.max(8, (int)(Runtime.getRuntime().availableProcessors() * 0.25));
+
   /**
    * The maximum allowed number of matrix transfer requests which are sending to the servers(ps). It
    * used to flow-control between psagent and ps.

diff --git a/...-ps/core/src/main/java/com/tencent/angel/psagent/matrix/oplog/cache/MatrixOpLogCache.java b/...-ps/core/src/main/java/com/tencent/angel/psagent/matrix/oplog/cache/MatrixOpLogCache.java
@@ -16,6 +16,7 @@
 
 package com.tencent.angel.psagent.matrix.oplog.cache;
 
+import com.tencent.angel.conf.AngelConf;
 import com.tencent.angel.conf.MatrixConf;
 import com.tencent.angel.ml.math.TUpdate;
 import com.tencent.angel.ml.matrix.MatrixMeta;
@@ -97,7 +98,10 @@ public MatrixOpLogCache() {
    * merge/flush tasks
    */
   public void start() {
-    workerPool = Executors.newCachedThreadPool();
+    workerPool = Executors.newFixedThreadPool(PSAgentContext.get().getConf().getInt(
+      AngelConf.ANGEL_MATRIX_OPLOG_MERGER_POOL_SIZE,
+      AngelConf.DEFAULT_ANGEL_MATRIX_OPLOG_MERGER_POOL_SIZE));
+
     dispatcher = new MergeDispacher();
     dispatcher.setName("oplog-merge-dispatcher");
     dispatcher.start();