Skip to content

Commit 6575c0c

Browse files
mgqa34mgqa34
authored
Merge pull request #560 from WeBankFinTech/develop
Develop
2 parents 253309c + fbca82a commit 6575c0c

26 files changed

Lines changed: 694 additions & 65 deletions

File tree

‎BUILD_INFO‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
eggroll.version=2.3.1
1+
eggroll.version=2.4.2

‎bin/roll_pair/egg_pair_bootstrap.sh‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ ONE_ARG_LIST=(
2222
"processor-id"
2323
"port"
2424
"transfer-port"
25+
"python-path"
2526
)
2627

2728
get_property() {
@@ -64,6 +65,14 @@ while [[ $# -gt 0 ]]; do
6465
server_node_id=$2
6566
shift 2
6667
;;
68+
--python-path)
69+
python_path=$2
70+
shift 2
71+
;;
72+
--python-venv)
73+
venv=$2
74+
shift 2
75+
;;
6776
*)
6877
break
6978
;;
@@ -85,8 +94,12 @@ if [[ ${transfer_port} -eq 0 ]] && [[ ${port} -ne 0 ]]; then
8594
transfer_port=${port}
8695
fi
8796

88-
get_property ${config} "eggroll.resourcemanager.bootstrap.egg_pair.venv"
89-
venv=${property_value}
97+
if [ ! $venv ]; then
98+
get_property ${config} "eggroll.resourcemanager.bootstrap.egg_pair.venv"
99+
venv=${property_value}
100+
else
101+
echo "venv defined=${venv}"
102+
fi
90103

91104
get_property ${config} "eggroll.resourcemanager.bootstrap.egg_pair.pythonpath"
92105
pythonpath=${property_value}
@@ -159,7 +172,8 @@ export MALLOC_MMAP_THRESHOLD_=${malloc_mmap_threshold}
159172
echo "MALLOC_MMAP_THRESHOLD_=${MALLOC_MMAP_THRESHOLD_}"
160173
export MALLOC_MMAP_MAX_=${malloc_mmap_max}
161174
echo "MALLOC_MMAP_MAX_=${MALLOC_MMAP_MAX_}"
162-
export PYTHONPATH=${pythonpath}:${PYTHONPATH}
175+
export PYTHONPATH=${python_path}:${pythonpath}:${PYTHONPATH}
176+
echo "python_path=${python_path}"
163177
echo "PYTHONPATH=${PYTHONPATH}"
164178
echo "PYTHON=`which python`"
165179

‎jvm/core/main/scala/com/webank/eggroll/core/constant/ConfKeys.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,8 @@ object SessionConfKeys {
148148
val CONFKEY_SESSION_PROCESSORS_PER_NODE = "eggroll.session.processors.per.node"
149149
val EGGROLL_SESSION_START_TIMEOUT_MS = ErConfKey("eggroll.session.start.timeout.ms", 20000)
150150
val EGGROLL_SESSION_STOP_TIMEOUT_MS = ErConfKey("eggroll.session.stop.timeout.ms", 20000)
151+
val EGGROLL_SESSION_PYTHON_PATH = "python.path"
152+
val EGGROLL_SESSION_PYTHON_VENV = "python.venv"
151153
}
152154

153155
object DeployConfKeys {

‎jvm/core/main/scala/com/webank/eggroll/core/resourcemanager/Container.scala‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,13 +44,34 @@ class Container(conf: RuntimeErConf, moduleName: String, processorId: Long = 0)
4444
private val boot = conf.getString(CoreConfKeys.BOOTSTRAP_ROOT_SCRIPT, s"bin/eggroll_boot.${if(isWindows) "py" else "sh"}")
4545
private val logsDir = s"${CoreConfKeys.EGGROLL_LOGS_DIR.get()}"
4646
private val cmPort = conf.getString(ClusterManagerConfKeys.CONFKEY_CLUSTER_MANAGER_PORT)
47+
private val pythonPath = conf.getString(SessionConfKeys.EGGROLL_SESSION_PYTHON_PATH)
48+
private val pythonVenv = conf.getString(SessionConfKeys.EGGROLL_SESSION_PYTHON_VENV)
4749

4850
if (StringUtils.isBlank(sessionId)) {
4951
throw new IllegalArgumentException("session Id is blank when creating processor")
5052
}
5153

5254
def start(): Boolean = {
53-
val startCmd = s"""${exeCmd} ${boot} start "${exePath} --config ${conf.getString(CoreConfKeys.STATIC_CONF_PATH)} --session-id ${sessionId} --server-node-id ${myServerNodeId} --processor-id ${processorId}" ${moduleName}-${processorId} &"""
55+
var startCmd = ""
56+
var pythonPathArgs = ""
57+
var pythonVenvArgs = ""
58+
59+
if (pythonPath.nonEmpty) {
60+
pythonPathArgs = s"--python-path ${pythonPath}"
61+
}
62+
63+
if (pythonVenv.nonEmpty) {
64+
pythonVenvArgs = s"--python-venv ${pythonVenv}"
65+
}
66+
67+
startCmd = s"""${exeCmd} ${boot} start "${exePath} --config ${conf.getString(CoreConfKeys.STATIC_CONF_PATH)} ${pythonPathArgs} ${pythonVenvArgs} --session-id ${sessionId} --server-node-id ${myServerNodeId} --processor-id ${processorId}" ${moduleName}-${processorId} &"""
68+
69+
// if (pythonPath.isEmpty) {
70+
// startCmd = s"""${exeCmd} ${boot} start "${exePath} --config ${conf.getString(CoreConfKeys.STATIC_CONF_PATH)} --session-id ${sessionId} --server-node-id ${myServerNodeId} --processor-id ${processorId}" ${moduleName}-${processorId} &"""
71+
// } else {
72+
// startCmd = s"""${exeCmd} ${boot} start "${exePath} --config ${conf.getString(CoreConfKeys.STATIC_CONF_PATH)} --python-path ${pythonPath} --session-id ${sessionId} --server-node-id ${myServerNodeId} --processor-id ${processorId}" ${moduleName}-${processorId} &"""
73+
// }
74+
5475
val standaloneTag = System.getProperty("eggroll.standalone.tag", StringConstants.EMPTY)
5576
logInfo(s"${standaloneTag} ${startCmd}")
5677

‎jvm/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -201,7 +201,7 @@
201201
<modelVersion>4.0.0</modelVersion>
202202

203203
<properties>
204-
<eggroll.version>2.3.1</eggroll.version>
204+
<eggroll.version>2.4.2</eggroll.version>
205205

206206
<!-- Languages -->
207207
<code.cache.size>512m</code.cache.size>
@@ -216,15 +216,15 @@
216216
<!-- Dependencies -->
217217
<arrow.version>0.17.1</arrow.version>
218218
<commons-beanutils.version>1.9.4</commons-beanutils.version>
219-
<commons-io.version>2.6</commons-io.version>
219+
<commons-io.version>2.11.0</commons-io.version>
220220
<commons-lang3.version>3.8.1</commons-lang3.version>
221221
<guava.version>30.1.1-jre</guava.version>
222222
<grpc.version>1.27.2</grpc.version>
223223
<javax.annotation-api.version>1.3.2</javax.annotation-api.version>
224224
<junit4.version>4.13.2</junit4.version>
225225
<junit.jupiter.version>5.5.1</junit.jupiter.version>
226226
<junit.platform.version>1.5.1</junit.platform.version>
227-
<log4j2.version>2.13.3</log4j2.version>
227+
<log4j2.version>2.17.1</log4j2.version>
228228
<netty.version>4.1.63.Final</netty.version>
229229
<protobuf.version>3.6.1</protobuf.version>
230230
<rocksdb.version>6.8.1</rocksdb.version>

‎jvm/roll_frame/pom.xml‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,21 @@
5555
<version>${spark.version}</version>
5656
<scope>provided</scope>
5757
</dependency>
58+
<dependency>
59+
<groupId>org.apache.arrow</groupId>
60+
<artifactId>arrow-memory</artifactId>
61+
<version>${arrow.version}</version>
62+
<exclusions>
63+
<exclusion>
64+
<groupId>io.netty</groupId>
65+
<artifactId>netty-buffer</artifactId>
66+
</exclusion>
67+
<exclusion>
68+
<groupId>io.netty</groupId>
69+
<artifactId>netty-common</artifactId>
70+
</exclusion>
71+
</exclusions>
72+
</dependency>
5873
<dependency>
5974
<groupId>com.fasterxml.jackson.core</groupId>
6075
<artifactId>jackson-core</artifactId>

‎jvm/roll_site/main/scala/com/webank/eggroll/rollsite/EggSiteServicer.scala‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,10 +104,12 @@ class EggSiteServicer extends DataTransferServiceGrpc.DataTransferServiceImplBas
104104

105105
var pollingExchanger: PollingExchanger = null
106106
var i = 0
107+
val partyId = dstPartyId
107108
while (pollingExchanger == null) {
108-
pollingExchanger = PollingExchanger.pollingExchangerQueue.poll(
109+
110+
pollingExchanger = PollingExchanger.getPollingExchangerQueue(partyId).poll(
109111
RollSiteConfKeys.EGGROLL_ROLLSITE_POLLING_Q_POLL_INTERVAL_SEC.get().toLong, TimeUnit.SECONDS)
110-
logTrace(s"unary call getting pollingExchanger from queue. i=${i}, isNull=${pollingExchanger == null}")
112+
logTrace(s"unary call getting pollingExchanger from queue. partyId=${partyId}, i=${i}, isNull=${pollingExchanger == null}")
111113
i += 1
112114
}
113115
pollingExchanger.setMethod(PollingMethods.UNARY_CALL)

‎jvm/roll_site/main/scala/com/webank/eggroll/rollsite/Polling.scala‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,20 @@ class PollingExchanger() {
216216
}
217217

218218
object PollingExchanger extends Logging {
219-
val pollingExchangerQueue = new LinkedBlockingQueue[PollingExchanger]()
219+
val pollingExchangerQueueMap = new java.util.concurrent.ConcurrentHashMap[String, LinkedBlockingQueue[PollingExchanger]]()
220+
221+
def getPollingExchangerQueue(partyId: String): LinkedBlockingQueue[PollingExchanger] = {
222+
if (!pollingExchangerQueueMap.containsKey(partyId)) {
223+
this.synchronized{
224+
if (!pollingExchangerQueueMap.containsKey(partyId)) {
225+
val pollingExchangerQueue = new LinkedBlockingQueue[PollingExchanger]()
226+
pollingExchangerQueueMap.put(partyId, pollingExchangerQueue)
227+
logDebug(s"created pollingExchangerQueue for party=${partyId}")
228+
}
229+
}
230+
}
231+
pollingExchangerQueueMap.get(partyId)
232+
}
220233

221234
def offer(data: Proxy.PollingFrame, q: SynchronousQueue[Proxy.PollingFrame], logPrefix: String, rsHeader: ErRollSiteHeader = null, metadataString: String = null): Boolean = {
222235
var done = false
@@ -336,17 +349,18 @@ class DispatchPollingReqSO(eggSiteServicerPollingRespSO: ServerCallStreamObserve
336349
pollingExchanger = new PollingExchanger()
337350
var done = false
338351
var i = 0
352+
val partyId = req.getMetadata.getDst.getPartyId
339353
val exchangerDataOpTimeout = System.currentTimeMillis() + RollSiteConfKeys.EGGROLL_ROLLSITE_POLLING_EXCHANGER_DATA_OP_TIMEOUT_SEC.get().toLong * 1000
340354
while (!done && System.currentTimeMillis() < exchangerDataOpTimeout) {
341-
done = PollingExchanger.pollingExchangerQueue.offer(pollingExchanger,
355+
done = PollingExchanger.getPollingExchangerQueue(partyId).offer(pollingExchanger,
342356
RollSiteConfKeys.EGGROLL_ROLLSITE_POLLING_Q_OFFER_INTERVAL_SEC.get().toLong, TimeUnit.SECONDS)
343-
logTrace(s"DispatchPollingReqSO.ensureInited calling, getting from pollingExchangerQueue. i=${i}")
357+
logTrace(s"DispatchPollingReqSO.ensureInited calling, getting from pollingExchangerQueue. partyId=${partyId}, i=${i}")
344358
i += 1
345359
}
346360

347361
if (!done) {
348-
onError(new TimeoutException("timeout when offering pollingExchanger to queue"))
349-
PollingExchanger.pollingExchangerQueue.remove(pollingExchanger)
362+
onError(new TimeoutException(s"timeout when offering pollingExchanger to queue, partyId=${partyId}"))
363+
PollingExchanger.getPollingExchangerQueue(partyId).remove(pollingExchanger)
350364
return
351365
}
352366
// synchronise point for incoming push / unary_call request
@@ -369,8 +383,8 @@ class DispatchPollingReqSO(eggSiteServicerPollingRespSO: ServerCallStreamObserve
369383
case PollingMethods.MOCK =>
370384
delegateSO = new MockPollingReqSO(eggSiteServicerPollingRespSO)
371385
case null =>
372-
PollingExchanger.pollingExchangerQueue.remove(pollingExchanger)
373-
throw new CancellationException("timeout in waiting polling method")
386+
PollingExchanger.getPollingExchangerQueue(partyId).remove(pollingExchanger)
387+
throw new CancellationException(s"timeout in waiting polling method, partyId=${partyId}")
374388
case _ =>
375389
val e = new NotImplementedError(s"method ${method} not supported")
376390
logError(e)

‎jvm/roll_site/main/scala/com/webank/eggroll/rollsite/Push.scala‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -317,9 +317,10 @@ class ForwardPushToPollingReqSO(eggSiteServicerPushRespSO_forwardPushToPollingRe
317317
logTrace(s"ForwardPushToPollingReqSO.ensureInited calling. rsKey=${rsKey}, rsHeader=${rsHeader}, metadata=${oneLineStringMetadata}")
318318

319319
while (pollingExchanger == null) {
320-
pollingExchanger = PollingExchanger.pollingExchangerQueue.poll(
320+
val partyId = firstRequest.getHeader.getDst.getPartyId
321+
pollingExchanger = PollingExchanger.getPollingExchangerQueue(partyId).poll(
321322
RollSiteConfKeys.EGGROLL_ROLLSITE_POLLING_Q_POLL_INTERVAL_SEC.get().toLong, TimeUnit.SECONDS)
322-
logTrace(s"ForwardPushToPollingReqSO.ensureInited, polling from pollingExchanger. isNull=${pollingExchanger == null}, rsKey=${rsKey}, rsHeader=${rsHeader}, metadata=${oneLineStringMetadata}")
323+
logTrace(s"ForwardPushToPollingReqSO.ensureInited, polling from pollingExchanger. partyId=${partyId}, isNull=${pollingExchanger == null}, rsKey=${rsKey}, rsHeader=${rsHeader}, metadata=${oneLineStringMetadata}")
323324
}
324325

325326
metadata = firstRequest.getHeader

‎jvm/roll_site/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,14 +33,14 @@
3333
</parent>
3434

3535
<properties>
36-
<spring.version>5.1.14.RELEASE</spring.version>
36+
<spring.version>5.1.20.RELEASE</spring.version>
3737
</properties>
3838

3939
<dependencies>
4040
<dependency>
4141
<groupId>org.apache.httpcomponents</groupId>
4242
<artifactId>httpclient</artifactId>
43-
<version>4.5.7</version>
43+
<version>4.5.13</version>
4444
</dependency>
4545

4646
<dependency>

0 commit comments

Comments
 (0)