bitflicker64 commented on code in PR #814:
URL:
https://github.com/apache/hugegraph-toolchain/pull/814#discussion_r4275949489
##########
hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/LoadTaskService.java:
##########
@@ -231,39 +410,93 @@ public List<LoadTask> batchTasks(int jobId) {
public LoadTask start(GraphConnection connection, FileMapping fileMapping,
HugeClient client) {
- LoadTask task = this.buildLoadTask(connection, fileMapping, client);
- this.save(task);
- this.executeAfterCommit(task);
- return task;
+ synchronized (this.admissionLock) {
+ this.checkAcceptingLoads();
+ }
+ Integer jobId = fileMapping.getJobId();
+ Ex.check(jobId != null, "File mapping has no job id");
+ try (TaskReservation reservation = this.reserveJob(jobId)) {
+ this.ensureParentExists(jobId, connection.getGraphSpace(),
connection.getGraph());
+ Ex.check(Objects.equals(fileMapping.getGraphSpace(),
connection.getGraphSpace()) &&
+ Objects.equals(fileMapping.getGraph(),
connection.getGraph()),
+ "File mapping does not belong to the load task graph");
+ FileMapping current =
this.fileMappingMapper.selectById(fileMapping.getId());
+ Ex.check(current != null && Objects.equals(current.getJobId(),
jobId) &&
+ Objects.equals(current.getGraphSpace(),
connection.getGraphSpace()) &&
+ Objects.equals(current.getGraph(), connection.getGraph()),
+ "File mapping does not belong to the load task job");
+ synchronized (this.admissionLock) {
+ this.checkAcceptingLoads();
+ LoadTask task = this.buildLoadTask(connection, current,
client);
Review Comment:
Minor: `start()` now builds the task inside `synchronized
(this.admissionLock)`. `buildLoadTask` reads the schema over HTTP
(`vlService.get` and `elService.get`) and creates the Loader's clients, which
also checks the Server version. `resume` and `retry` do the same under this
lock through `task.reconnect(token)`, which builds a new `HugeGraphLoader`.
There is one lock per service, shared by all graphs.
While one of these waits on a slow Server (the schema client's
`client.request_timeout` defaults to 310 s), every other user of the lock waits
too: other starts, resume and retry, delete (`requestJobStopForDelete`,
`checkCallbackWaitAllowed`), `pauseAllTasks` at shutdown, and the callback of
every finished import (`complete()` -> `release()`), which also keeps its
executor thread. The base built the task without any lock.
Please build the Loader before taking the lock, keep only the admission
check, the metadata write and the dispatch inside it, and close the unused
Loader if admission is refused.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
