hugegraph
/
hugegraph-sync
mirror of https://github.com/apache/hugegraph
0
0
Fork 2
Code Issues Packages Projects Releases Wiki Activity

fix(server): avoid loading huge task results for metadata queries (#3060)

This commit is contained in:
contrueCT 2026-06-21 20:27:45 +08:00 committed by GitHub
parent c3f56b5e9f
commit 63c97ff93f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 778 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
View File

@ -102,12 +102,13 @@ public class TaskAPI extends API {
limit = NO_LIMIT;
List<Id> idList = ids.stream().map(IdGenerator::of)
.collect(Collectors.toList());
iter = scheduler.tasks(idList);
iter = scheduler.tasks(idList, false);
} else {
if (status == null) {
iter = scheduler.tasks(null, limit, page);
iter = scheduler.tasks(null, limit, page, false);
} else {
iter = scheduler.tasks(parseStatus(status), limit, page);
iter = scheduler.tasks(parseStatus(status), limit, page,
false);
}
}
@ -136,12 +137,17 @@ public class TaskAPI extends API {
@Parameter(description = "The graph name")
@PathParam("graph") String graph,
@Parameter(description = "The task id")
@PathParam("id") long id) {
@PathParam("id") long id,
@Parameter(description = "Whether to load task result")
@DefaultValue("true")
@QueryParam("with_result")
boolean withResult) {
LOG.debug("Graph [{}] get task: {}", graph, id);
TaskScheduler scheduler = graph(manager, graphSpace, graph)
.taskScheduler();
return scheduler.task(IdGenerator.of(id)).asMap();
return scheduler.task(IdGenerator.of(id), withResult)
.asMap(true, withResult);
}
@DELETE

30
hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
View File

@ -1326,8 +1326,14 @@ public final class HugeGraphAuthProxy implements HugeGraph {
@Override
public <V> HugeTask<V> task(Id id) {
return this.task(id, true);
}
@Override
public <V> HugeTask<V> task(Id id, boolean withResult) {
return verifyTaskPermission(HugePermission.READ,
this.taskScheduler.task(id));
this.taskScheduler.task(id,
withResult));
}
@Override
@ -1336,18 +1342,36 @@ public final class HugeGraphAuthProxy implements HugeGraph {
this.taskScheduler.tasks(ids));
}
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
boolean withResult) {
return verifyTaskPermission(HugePermission.READ,
this.taskScheduler.tasks(ids,
withResult));
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
long limit, String page) {
Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(status,
limit, page);
limit,
page);
return verifyTaskPermission(HugePermission.READ, tasks);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
long limit, String page,
boolean withResult) {
Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(
status, limit, page, withResult);
return verifyTaskPermission(HugePermission.READ, tasks);
}
@Override
public <V> HugeTask<V> delete(Id id, boolean force) {
verifyTaskPermission(HugePermission.DELETE,
this.taskScheduler.task(id));
this.taskScheduler.task(id, false));
return this.taskScheduler.delete(id, force);
}

18
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java
View File

@ -306,10 +306,14 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
if (vertex == null) {
this.deleteTaskResultFromTx(id);
return null;
}
HugeTask<V> result = HugeTask.fromVertex(vertex);
this.tx().removeVertex(vertex);
HugeTask<V> result = HugeTask.fromVertex(vertex, false);
// Keep the task vertex as a retryable tombstone until its result
// vertex is removed; cronSchedule() can rediscover DELETING tasks.
this.deleteTaskResultFromTx(id);
this.tx().removeTaskVertex(vertex);
return result;
});
}
@ -322,6 +326,12 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler {
this.updateStatus(id, null, TaskStatus.DELETING);
return null;
} else {
HugeTask<?> task = this.taskWithoutResult(id);
if (task != null && task.status() != TaskStatus.DELETING) {
initTaskParams(task);
task.overwriteStatus(TaskStatus.DELETING);
this.save(task);
}
return this.deleteFromDB(id);
}
}
@ -587,7 +597,7 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler {
}
}
private boolean isLockedTask(String taskId) {
protected boolean isLockedTask(String taskId) {
return MetaManager.instance().isLockedTask(graphSpace,
graphName, taskId);
}
@ -629,7 +639,7 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler {
// 1. start task can be from schedule() & cronSchedule()
// 2. recheck the status of task, in case one same task
// called by both methods at same time;
HugeTask<Object> queryTask = task(this.task.id());
HugeTask<Object> queryTask = task(this.task.id(), false);
if (queryTask != null &&
!TaskStatus.NEW.equals(queryTask.status())) {
return;

69
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java
View File

@ -41,6 +41,7 @@ import org.apache.hugegraph.job.ComputerJob;
import org.apache.hugegraph.job.EphemeralJob;
import org.apache.hugegraph.job.GremlinJob;
import org.apache.hugegraph.job.schema.SchemaJob;
import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.type.define.SerialEnum;
import org.apache.hugegraph.util.Blob;
import org.apache.hugegraph.util.E;
@ -653,6 +654,11 @@ public class HugeTask<V> extends FutureTask<V> {
}
public synchronized Map<String, Object> asMap(boolean withDetails) {
return this.asMap(withDetails, true);
}
public synchronized Map<String, Object> asMap(boolean withDetails,
boolean withResult) {
E.checkState(this.type != null, "Task type can't be null");
E.checkState(this.name != null, "Task name can't be null");
@ -689,7 +695,7 @@ public class HugeTask<V> extends FutureTask<V> {
if (this.input != null) {
map.put(Hidden.unHide(P.INPUT), this.input);
}
if (this.result != null) {
if (withResult && this.result != null) {
map.put(Hidden.unHide(P.RESULT), this.result);
}
}
@ -697,7 +703,37 @@ public class HugeTask<V> extends FutureTask<V> {
return map;
}
synchronized HugeTask<V> copyWithoutResult() {
HugeTask<V> task = new HugeTask<>(this.id, this.parent, this.callable);
task.type = this.type;
task.name = this.name;
task.dependencies = this.dependencies == null ?
null : InsertionOrderUtil.newSet(this.dependencies);
task.description = this.description;
task.context = this.context;
task.create = this.create;
task.server = this.server;
task.load = this.load;
task.status = this.status;
task.progress = this.progress;
task.update = this.update;
task.retries = this.retries;
task.input = this.input;
task.result = null;
task.scheduler = this.scheduler;
return task;
}
public static <V> HugeTask<V> fromVertex(Vertex vertex) {
return fromVertex(vertex, true);
}
public static <V> HugeTask<V> fromVertex(Vertex vertex,
boolean withResult) {
if (!withResult && vertex instanceof HugeVertex) {
return fromHugeVertex((HugeVertex) vertex);
}
String callableName = vertex.value(P.CALLABLE);
TaskCallable<V> callable;
try {
@ -710,11 +746,37 @@ public class HugeTask<V> extends FutureTask<V> {
for (Iterator<VertexProperty<Object>> iter = vertex.properties();
iter.hasNext(); ) {
VertexProperty<Object> prop = iter.next();
if (!withResult && P.RESULT.equals(prop.key())) {
continue;
}
task.property(prop.key(), prop.value());
}
return task;
}
private static <V> HugeTask<V> fromHugeVertex(HugeVertex vertex) {
String callableName = getPropertyValue(vertex, P.CALLABLE);
TaskCallable<V> callable;
try {
callable = TaskCallable.fromClass(callableName);
} catch (Exception e) {
callable = TaskCallable.empty(e);
}
HugeTask<V> task = new HugeTask<>(vertex.id(), null, callable);
for (String property : P.METADATA_PROPERTIES) {
Object value = getPropertyValue(vertex, property);
if (value != null) {
task.property(property, value);
}
}
return task;
}
private static <V> V getPropertyValue(HugeVertex vertex, String property) {
return vertex.getPropertyValue(vertex.graph().propertyKey(property).id());
}
private static <V> Collector<V, ?, Set<V>> toOrderSet() {
return Collectors.toCollection(InsertionOrderUtil::newSet);
}
@ -792,6 +854,11 @@ public class HugeTask<V> extends FutureTask<V> {
public static final String DEPENDENCIES = "~task_dependencies";
public static final String SERVER = "~task_server";
private static final String[] METADATA_PROPERTIES = new String[]{
TYPE, NAME, CALLABLE, DESCRIPTION, CONTEXT, STATUS, PROGRESS,
CREATE, UPDATE, RETRIES, DEPENDENCIES, INPUT, SERVER
};
//public static final String PARENT = hide("parent");
//public static final String CHILDREN = hide("children");

107
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
View File

@ -153,7 +153,7 @@ public class StandardTaskScheduler implements TaskScheduler {
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
Iterator<HugeTask<V>> iter;
for (iter = this.findTask(status, PAGE_SIZE, page);
for (iter = this.findTask(status, PAGE_SIZE, page, false);
iter.hasNext(); ) {
HugeTask<V> task = iter.next();
if (selfServer.equals(task.server())) {
@ -323,7 +323,8 @@ public class StandardTaskScheduler implements TaskScheduler {
Collection<HugeServerInfo> serverInfos = this.serverManager().allServerInfos();
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.SCHEDULING, PAGE_SIZE, page);
Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.SCHEDULING, PAGE_SIZE, page,
false);
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
if (task.server() != null) {
@ -365,7 +366,8 @@ public class StandardTaskScheduler implements TaskScheduler {
protected void executeTasksOnWorker(Id server) {
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.SCHEDULED, PAGE_SIZE, page);
Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.SCHEDULED, PAGE_SIZE, page,
false);
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
this.initTaskCallable(task);
@ -394,7 +396,8 @@ public class StandardTaskScheduler implements TaskScheduler {
protected void cancelTasksOnWorker(Id server) {
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.CANCELLING, PAGE_SIZE, page);
Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.CANCELLING, PAGE_SIZE, page,
false);
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
Id taskServer = task.server();
@ -494,24 +497,35 @@ public class StandardTaskScheduler implements TaskScheduler {
@Override
public <V> HugeTask<V> task(Id id) {
return this.task(id, true);
}
@Override
public <V> HugeTask<V> task(Id id, boolean withResult) {
E.checkArgumentNotNull(id, "Parameter task id can't be null");
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
if (task != null) {
return task;
return withResult ? task : task.copyWithoutResult();
}
return this.findTask(id);
return this.findTask(id, withResult);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
return this.tasks(ids, true);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
boolean withResult) {
List<Id> taskIdsNotInMem = new ArrayList<>();
List<HugeTask<V>> taskInMem = new ArrayList<>();
for (Id id : ids) {
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
if (task != null) {
taskInMem.add(task);
taskInMem.add(withResult ? task : task.copyWithoutResult());
} else {
taskIdsNotInMem.add(id);
}
@ -522,27 +536,37 @@ public class StandardTaskScheduler implements TaskScheduler {
} else {
iterator = new ExtendableIterator<>(taskInMem.iterator());
}
iterator.extend(this.findTasks(taskIdsNotInMem));
iterator.extend(this.findTasks(taskIdsNotInMem, withResult));
return iterator;
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
long limit, String page) {
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
String page) {
return this.tasks(status, limit, page, true);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
String page, boolean withResult) {
if (status == null) {
return this.findAllTask(limit, page);
return this.findAllTask(limit, page, withResult);
}
return this.findTask(status, limit, page);
return this.findTask(status, limit, page, withResult);
}
public <V> HugeTask<V> findTask(Id id) {
return this.findTask(id, true);
}
public <V> HugeTask<V> findTask(Id id, boolean withResult) {
HugeTask<V> result = this.call(() -> {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
Vertex vertex = QueryResults.one(vertices);
if (vertex == null) {
return null;
}
return HugeTask.fromVertex(vertex);
return HugeTask.fromVertex(vertex, withResult);
});
if (result == null) {
throw new NotFoundException("Can't find task with id '%s'", id);
@ -551,23 +575,40 @@ public class StandardTaskScheduler implements TaskScheduler {
}
public <V> Iterator<HugeTask<V>> findTasks(List<Id> ids) {
return this.queryTask(ids);
return this.findTasks(ids, true);
}
public <V> Iterator<HugeTask<V>> findTasks(List<Id> ids,
boolean withResult) {
return this.queryTask(ids, withResult);
}
public <V> Iterator<HugeTask<V>> findAllTask(long limit, String page) {
return this.queryTask(ImmutableMap.of(), limit, page);
return this.findAllTask(limit, page, true);
}
public <V> Iterator<HugeTask<V>> findAllTask(long limit, String page,
boolean withResult) {
return this.queryTask(ImmutableMap.of(), limit, page, withResult);
}
public <V> Iterator<HugeTask<V>> findTask(TaskStatus status,
long limit, String page) {
return this.queryTask(P.STATUS, status.code(), limit, page);
return this.findTask(status, limit, page, true);
}
public <V> Iterator<HugeTask<V>> findTask(TaskStatus status,
long limit, String page,
boolean withResult) {
return this.queryTask(P.STATUS, status.code(), limit, page,
withResult);
}
@Override
public <V> HugeTask<V> delete(Id id, boolean force) {
this.checkOnMasterNode("delete");
HugeTask<?> task = this.task(id);
HugeTask<?> task = this.task(id, false);
/*
* The following is out of date when the task running on worker node:
* HugeTask<?> task = this.tasks.get(id);
@ -592,11 +633,11 @@ public class StandardTaskScheduler implements TaskScheduler {
if (vertex == null) {
return null;
}
HugeTask<V> result = HugeTask.fromVertex(vertex);
HugeTask<V> result = HugeTask.fromVertex(vertex, false);
E.checkState(force || result.completed(),
"Can't delete incomplete task '%s' in status %s",
id, result.status());
this.tx().removeVertex(vertex);
this.tx().removeTaskVertex(vertex);
return result;
});
}
@ -674,11 +715,24 @@ public class StandardTaskScheduler implements TaskScheduler {
private <V> Iterator<HugeTask<V>> queryTask(String key, Object value,
long limit, String page) {
return this.queryTask(ImmutableMap.of(key, value), limit, page);
return this.queryTask(key, value, limit, page, true);
}
private <V> Iterator<HugeTask<V>> queryTask(String key, Object value,
long limit, String page,
boolean withResult) {
return this.queryTask(ImmutableMap.of(key, value), limit, page,
withResult);
}
private <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
long limit, String page) {
return this.queryTask(conditions, limit, page, true);
}
private <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
long limit, String page,
boolean withResult) {
return this.call(() -> {
ConditionQuery query;
if (this.graph.backendStoreFeatures().supportsTaskAndServerVertex()) {
@ -701,18 +755,27 @@ public class StandardTaskScheduler implements TaskScheduler {
}
Iterator<Vertex> vertices = this.tx().queryVertices(query);
Iterator<HugeTask<V>> tasks =
new MapperIterator<>(vertices, HugeTask::fromVertex);
new MapperIterator<>(vertices, vertex -> {
return HugeTask.fromVertex(vertex, withResult);
});
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
}
private <V> Iterator<HugeTask<V>> queryTask(List<Id> ids) {
return this.queryTask(ids, true);
}
private <V> Iterator<HugeTask<V>> queryTask(List<Id> ids,
boolean withResult) {
return this.call(() -> {
Object[] idArray = ids.toArray(new Id[0]);
Iterator<Vertex> vertices = this.tx().queryTaskInfos(idArray);
Iterator<HugeTask<V>> tasks =
new MapperIterator<>(vertices, HugeTask::fromVertex);
new MapperIterator<>(vertices, vertex -> {
return HugeTask.fromVertex(vertex, withResult);
});
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});

66
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java
View File

@ -104,22 +104,29 @@ public abstract class TaskAndResultScheduler implements TaskScheduler {
@Override
public <V> HugeTask<V> task(Id id) {
return this.task(id, true);
}
@Override
public <V> HugeTask<V> task(Id id, boolean withResult) {
HugeTask<V> task = this.call(() -> {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
Vertex vertex = QueryResults.one(vertices);
if (vertex == null) {
return null;
}
return HugeTask.fromVertex(vertex);
return HugeTask.fromVertex(vertex, withResult);
});
if (task == null) {
throw new NotFoundException("Can't find task with id '%s'", id);
}
HugeTaskResult taskResult = queryTaskResult(id);
if (taskResult != null) {
task.result(taskResult);
if (withResult) {
HugeTaskResult taskResult = queryTaskResult(id);
if (taskResult != null) {
task.result(taskResult);
}
}
return task;
@ -127,17 +134,39 @@ public abstract class TaskAndResultScheduler implements TaskScheduler {
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
return this.tasksWithoutResult(ids);
return this.tasks(ids, false);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
boolean withResult) {
if (!withResult) {
return this.tasksWithoutResult(ids);
}
return this.queryTask(ids);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
String page) {
if (status == null) {
return this.queryTaskWithoutResult(ImmutableMap.of(), limit, page);
return this.tasks(status, limit, page, false);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
String page, boolean withResult) {
if (!withResult) {
if (status == null) {
return this.queryTaskWithoutResult(ImmutableMap.of(), limit,
page);
}
return this.queryTaskWithoutResult(HugeTask.P.STATUS,
status.code(), limit, page);
}
return this.queryTaskWithoutResult(HugeTask.P.STATUS, status.code(),
limit, page);
if (status == null) {
return this.queryTask(ImmutableMap.of(), limit, page);
}
return this.queryTask(HugeTask.P.STATUS, status.code(), limit, page);
}
protected <V> Iterator<HugeTask<V>> queryTask(String key, Object value,
@ -216,7 +245,7 @@ public abstract class TaskAndResultScheduler implements TaskScheduler {
if (vertex == null) {
return null;
}
return HugeTask.fromVertex(vertex);
return HugeTask.fromVertex(vertex, false);
});
return result;
@ -227,7 +256,9 @@ public abstract class TaskAndResultScheduler implements TaskScheduler {
Object[] idArray = ids.toArray(new Id[ids.size()]);
Iterator<Vertex> vertices = this.tx().queryTaskInfos(idArray);
Iterator<HugeTask<V>> tasks =
new MapperIterator<>(vertices, HugeTask::fromVertex);
new MapperIterator<>(vertices, vertex -> {
return HugeTask.fromVertex(vertex, false);
});
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
@ -268,12 +299,23 @@ public abstract class TaskAndResultScheduler implements TaskScheduler {
}
Iterator<Vertex> vertices = this.tx().queryTaskInfos(query);
Iterator<HugeTask<V>> tasks =
new MapperIterator<>(vertices, HugeTask::fromVertex);
new MapperIterator<>(vertices, vertex -> {
return HugeTask.fromVertex(vertex, false);
});
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
}
protected void deleteTaskResultFromTx(Id taskId) {
Iterator<Vertex> vertices =
this.tx().queryTaskInfos(HugeTaskResult.genId(taskId));
HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
if (vertex != null) {
this.tx().removeTaskVertex(vertex);
}
}
protected HugeTaskResult queryTaskResult(Id taskid) {
HugeTaskResult result = this.call(() -> {
Iterator<Vertex> vertices =

14
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskScheduler.java
View File

@ -49,11 +49,25 @@ public interface TaskScheduler {
<V> HugeTask<V> task(Id id);
default <V> HugeTask<V> task(Id id, boolean withResult) {
return this.task(id);
}
<V> Iterator<HugeTask<V>> tasks(List<Id> ids);
default <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
boolean withResult) {
return this.tasks(ids);
}
<V> Iterator<HugeTask<V>> tasks(TaskStatus status,
long limit, String page);
default <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
String page, boolean withResult) {
return this.tasks(status, limit, page);
}
void init();
boolean close();

29
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskTransaction.java
View File

@ -31,6 +31,7 @@ import org.apache.hugegraph.schema.IndexLabel;
import org.apache.hugegraph.schema.PropertyKey;
import org.apache.hugegraph.schema.SchemaManager;
import org.apache.hugegraph.schema.VertexLabel;
import org.apache.hugegraph.structure.HugeIndex;
import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.Cardinality;
@ -74,6 +75,34 @@ public class TaskTransaction extends GraphTransaction {
return false;
}
public void removeTaskVertex(HugeVertex vertex) {
this.checkOwnerThread();
this.beforeWrite();
this.doRemove(this.serializer.writeVertex(vertex.prepareRemoved()));
if (TASK.equals(vertex.schemaLabel().name())) {
this.updateIndex(this.indexLabel(HugeTask.P.STATUS).id(),
vertex, true);
}
this.removeLabelIndex(vertex);
this.afterWrite();
}
private void removeLabelIndex(HugeVertex vertex) {
if (this.store().features().supportsQueryByLabel() ||
!vertex.schemaLabel().enableLabelIndex()) {
return;
}
HugeIndex index = new HugeIndex(this.graph(),
IndexLabel.label(vertex.type()));
index.fieldValues(vertex.schemaLabel().id());
index.elementIds(vertex.id(), vertex.expiredTime());
this.doEliminate(this.serializer.writeIndex(index));
}
public void initSchema() {
if (this.existVertexLabel(TASK)) {
return;

28
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/TaskApiTest.java
View File

@ -87,6 +87,28 @@ public class TaskApiTest extends BaseApiTest {
Assert.assertEquals("success", status);
}
@Test
public void testGetWithoutResult() {
int taskId = this.gremlinJob("1 + 2");
waitTaskSuccess(taskId);
Response r = client().get(PATH, ImmutableMap.of("limit", -1));
String content = assertResponseStatus(200, r);
Assert.assertFalse(content, content.contains("task_result"));
r = client().get(PATH, String.valueOf(taskId));
content = assertResponseStatus(200, r);
assertJsonContains(content, "task_result");
r = client().get(PATH + taskId,
ImmutableMap.of("with_result", false));
content = assertResponseStatus(200, r);
assertJsonContains(content, "id");
assertJsonContains(content, "task_callable");
Assert.assertFalse(content, content.contains("task_result"));
}
@Test
public void testCancel() {
// create a task
@ -143,8 +165,12 @@ public class TaskApiTest extends BaseApiTest {
}
private int gremlinJob() {
return this.gremlinJob("Thread.sleep(1000L)");
}
private int gremlinJob(String gremlin) {
String body = "{" +
"\"gremlin\":\"Thread.sleep(1000L)\"," +
"\"gremlin\":\"" + gremlin + "\"," +
"\"bindings\":{}," +
"\"language\":\"gremlin-groovy\"," +
"\"aliases\":{}}";

2
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java
View File

@ -23,6 +23,7 @@ import org.apache.hugegraph.dist.RegisterUtil;
import org.apache.hugegraph.masterelection.GlobalMasterInfo;
import org.apache.hugegraph.meta.MetaManager;
import org.apache.hugegraph.meta.PdMetaDriver;
import org.apache.hugegraph.task.TaskAndResultSchedulerTest;
import org.apache.hugegraph.testutil.Utils;
import org.apache.hugegraph.util.Log;
import org.junit.AfterClass;
@ -45,6 +46,7 @@ import org.slf4j.Logger;
PropertyCoreTest.EdgePropertyCoreTest.class,
RestoreCoreTest.class,
TaskCoreTest.class,
TaskAndResultSchedulerTest.class,
AuthTest.class,
MultiGraphsTest.class,
RamTableTest.class

99
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java
View File

@ -115,6 +115,88 @@ public class TaskCoreTest extends BaseCoreTest {
});
}
@Test
public void testTaskWithoutResult() throws TimeoutException {
HugeGraph graph = graph();
TaskScheduler scheduler = graph.taskScheduler();
Id id = IdGenerator.of(88889);
HugeTask<?> task = new HugeTask<>(id, null, new SleepCallable<>());
task.type("test");
task.name("metadata-task-in-memory");
scheduler.schedule(task);
try {
Whitebox.setInternalState(task, "result", "\"in-memory-result\"");
HugeTask<?> taskWithoutResult = scheduler.task(id, false);
Assert.assertEquals("metadata-task-in-memory",
taskWithoutResult.name());
Assert.assertNull(taskWithoutResult.result());
Iterator<HugeTask<Object>> iter = scheduler.tasks(ImmutableList.of(id),
false);
Assert.assertTrue(iter.hasNext());
taskWithoutResult = iter.next();
Assert.assertEquals("metadata-task-in-memory",
taskWithoutResult.name());
Assert.assertNull(taskWithoutResult.result());
Assert.assertFalse(iter.hasNext());
} finally {
Whitebox.setInternalState(task, "result", null);
}
scheduler.waitUntilTaskCompleted(id, 10);
scheduler.delete(id, false);
id = IdGenerator.of(88890);
task = new HugeTask<>(id, null, new MetadataResultCallable());
task.type("test");
task.name("metadata-task");
scheduler.schedule(task);
scheduler.waitUntilTaskCompleted(id, 10);
HugeTask<?> taskWithResult = scheduler.task(id, true);
Assert.assertEquals("\"metadata-result\"", taskWithResult.result());
HugeTask<?> taskWithoutResult = scheduler.task(id, false);
Assert.assertEquals("metadata-task", taskWithoutResult.name());
Assert.assertNull(taskWithoutResult.result());
Iterator<HugeTask<Object>> iter = scheduler.tasks(ImmutableList.of(id));
Assert.assertTrue(iter.hasNext());
taskWithResult = iter.next();
Assert.assertEquals("metadata-task", taskWithResult.name());
Assert.assertEquals("\"metadata-result\"", taskWithResult.result());
Assert.assertFalse(iter.hasNext());
iter = scheduler.tasks(ImmutableList.of(id), false);
Assert.assertTrue(iter.hasNext());
taskWithoutResult = iter.next();
Assert.assertEquals("metadata-task", taskWithoutResult.name());
Assert.assertNull(taskWithoutResult.result());
Assert.assertFalse(iter.hasNext());
iter = scheduler.tasks(TaskStatus.SUCCESS, 10, null, false);
Assert.assertTrue(iter.hasNext());
taskWithoutResult = iter.next();
Assert.assertEquals("metadata-task", taskWithoutResult.name());
Assert.assertNull(taskWithoutResult.result());
iter = scheduler.tasks(TaskStatus.SUCCESS, 10, null);
Assert.assertTrue(iter.hasNext());
taskWithResult = iter.next();
Assert.assertEquals("metadata-task", taskWithResult.name());
Assert.assertEquals("\"metadata-result\"", taskWithResult.result());
Id taskId = id;
scheduler.delete(taskId, false);
Assert.assertThrows(NotFoundException.class, () -> {
scheduler.task(taskId);
});
}
@Test
public void testTaskWithFailure() throws TimeoutException {
HugeGraph graph = graph();
@ -714,4 +796,21 @@ public class TaskCoreTest extends BaseCoreTest {
this.graph().taskScheduler().save(this.task());
}
}
public static class MetadataResultCallable extends TaskCallable<String> {
public MetadataResultCallable() {
// pass
}
@Override
public String call() {
return "metadata-result";
}
@Override
public void done() {
this.graph().taskScheduler().save(this.task());
}
}
}

348
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/task/TaskAndResultSchedulerTest.java Normal file
View File

@ -0,0 +1,348 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package org.apache.hugegraph.task;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hugegraph.HugeException;
import org.apache.hugegraph.HugeGraphParams;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.IdGenerator;
import org.apache.hugegraph.backend.query.QueryResults;
import org.apache.hugegraph.core.BaseCoreTest;
import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.testutil.Whitebox;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.junit.Test;
import com.google.common.collect.ImmutableList;
public class TaskAndResultSchedulerTest extends BaseCoreTest {
@Test
public void testMetadataOnlyReadsSkipLargeCompressedTaskResult() {
Id id = IdGenerator.of(88901);
TestDistributedTaskScheduler scheduler = this.newScheduler();
try {
scheduler.init();
String largeResult = largeTaskResult();
HugeTask<Object> task = newTask(id, "large-result-task",
TaskStatus.SUCCESS, largeResult);
scheduler.save(task);
Assert.assertTrue(scheduler.taskResultVertexExists(id));
HugeTask<?> taskWithResult = scheduler.task(id, true);
Assert.assertEquals(largeResult, taskWithResult.result());
Assert.assertGt(0, scheduler.resultReadCount());
scheduler.resetResultReadCount();
scheduler.forbidResultRead(true);
HugeTask<?> taskWithoutResult = scheduler.task(id, false);
assertTaskWithoutResult(taskWithoutResult, id);
Assert.assertEquals(0, scheduler.resultReadCount());
Iterator<HugeTask<Object>> tasks = scheduler.tasks(ImmutableList.of(id),
false);
assertIteratorContainsTaskWithoutResult(tasks, id);
Assert.assertEquals(0, scheduler.resultReadCount());
tasks = scheduler.tasks(TaskStatus.SUCCESS, 10L, null, false);
assertIteratorContainsTaskWithoutResult(tasks, id);
Assert.assertEquals(0, scheduler.resultReadCount());
tasks = scheduler.tasks(ImmutableList.of(id));
assertIteratorContainsTaskWithoutResult(tasks, id);
Assert.assertEquals(0, scheduler.resultReadCount());
tasks = scheduler.tasks(TaskStatus.SUCCESS, 10L, null);
assertIteratorContainsTaskWithoutResult(tasks, id);
Assert.assertEquals(0, scheduler.resultReadCount());
tasks = scheduler.queryMetadataTasksByStatus(TaskStatus.SUCCESS);
assertIteratorContainsTaskWithoutResult(tasks, id);
Assert.assertEquals(0, scheduler.resultReadCount());
scheduler.forbidResultRead(false);
scheduler.resetResultReadCount();
tasks = scheduler.tasks(ImmutableList.of(id), true);
assertIteratorContainsTaskWithResult(tasks, id, largeResult);
Assert.assertGt(0, scheduler.resultReadCount());
} finally {
scheduler.forbidResultRead(false);
scheduler.deleteFromDBForTest(id);
scheduler.closeAndShutdown();
}
}
@Test
public void testDistributedDeleteKeepsTaskResultRecoverable() {
Id id = IdGenerator.of(88902);
TestDistributedTaskScheduler scheduler = this.newScheduler();
try {
scheduler.init();
HugeTask<Object> task = newTask(id, "delete-order-task",
TaskStatus.SUCCESS, "\"result\"");
scheduler.save(task);
Assert.assertTrue(scheduler.taskVertexExists(id));
Assert.assertTrue(scheduler.taskResultVertexExists(id));
scheduler.failNextTaskResultDelete();
Assert.assertThrows(HugeException.class, () -> {
scheduler.delete(id, true);
});
Assert.assertTrue(scheduler.taskVertexExists(id));
Assert.assertEquals(TaskStatus.DELETING, scheduler.taskStatus(id));
Assert.assertTrue(scheduler.taskResultVertexExists(id));
scheduler.cronSchedule();
Assert.assertFalse(scheduler.taskVertexExists(id));
Assert.assertFalse(scheduler.taskResultVertexExists(id));
} finally {
scheduler.allowTaskResultDelete();
scheduler.deleteFromDBForTest(id);
scheduler.closeAndShutdown();
}
}
private TestDistributedTaskScheduler newScheduler() {
return new TestDistributedTaskScheduler(this.params());
}
private static HugeTask<Object> newTask(Id id, String name,
TaskStatus status, String result) {
HugeTask<Object> task = new HugeTask<>(id, null, new EmptyCallable());
task.type("test");
task.name(name);
task.overwriteStatus(status);
Whitebox.setInternalState(task, "result", result);
return task;
}
private static void assertTaskWithoutResult(HugeTask<?> task, Id id) {
Assert.assertEquals(id, task.id());
Assert.assertNull(task.result());
}
private static void assertIteratorContainsTaskWithoutResult(
Iterator<HugeTask<Object>> tasks, Id id) {
boolean matched = false;
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
if (id.equals(task.id())) {
assertTaskWithoutResult(task, id);
matched = true;
}
}
Assert.assertTrue(matched);
}
private static void assertIteratorContainsTaskWithResult(
Iterator<HugeTask<Object>> tasks, Id id, String result) {
boolean matched = false;
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
if (id.equals(task.id())) {
Assert.assertEquals(id, task.id());
Assert.assertEquals(result, task.result());
matched = true;
}
}
Assert.assertTrue(matched);
}
private static String largeTaskResult() {
char[] chars = new char[1024 * 1024 + 257];
int value = 0x12345678;
for (int i = 0; i < chars.length; i++) {
value = value * 1103515245 + 12345;
chars[i] = (char) ('a' + ((value >>> 16) & 0x0F));
}
return new String(chars);
}
public static class EmptyCallable extends TaskCallable<Object> {
public EmptyCallable() {
// pass
}
@Override
public Object call() {
return null;
}
}
private static class TestDistributedTaskScheduler
extends DistributedTaskScheduler {
private final ScheduledThreadPoolExecutor schedulerExecutor;
private final ExecutorService taskDbExecutor;
private final ExecutorService schemaTaskExecutor;
private final ExecutorService olapTaskExecutor;
private final ExecutorService gremlinTaskExecutor;
private final ExecutorService ephemeralTaskExecutor;
private final ExecutorService serverInfoDbExecutor;
private final AtomicInteger resultReadCount;
private volatile boolean forbidResultRead;
private volatile boolean failNextTaskResultDelete;
TestDistributedTaskScheduler(HugeGraphParams graph) {
this(graph, new ScheduledThreadPoolExecutor(1),
Executors.newSingleThreadExecutor(),
Executors.newSingleThreadExecutor(),
Executors.newSingleThreadExecutor(),
Executors.newSingleThreadExecutor(),
Executors.newSingleThreadExecutor(),
Executors.newSingleThreadExecutor());
}
private TestDistributedTaskScheduler(
HugeGraphParams graph,
ScheduledThreadPoolExecutor schedulerExecutor,
ExecutorService taskDbExecutor,
ExecutorService schemaTaskExecutor,
ExecutorService olapTaskExecutor,
ExecutorService gremlinTaskExecutor,
ExecutorService ephemeralTaskExecutor,
ExecutorService serverInfoDbExecutor) {
super(graph, schedulerExecutor, taskDbExecutor, schemaTaskExecutor,
olapTaskExecutor, gremlinTaskExecutor, ephemeralTaskExecutor,
serverInfoDbExecutor);
this.schedulerExecutor = schedulerExecutor;
this.taskDbExecutor = taskDbExecutor;
this.schemaTaskExecutor = schemaTaskExecutor;
this.olapTaskExecutor = olapTaskExecutor;
this.gremlinTaskExecutor = gremlinTaskExecutor;
this.ephemeralTaskExecutor = ephemeralTaskExecutor;
this.serverInfoDbExecutor = serverInfoDbExecutor;
this.resultReadCount = new AtomicInteger();
this.forbidResultRead = false;
this.failNextTaskResultDelete = false;
}
@Override
protected HugeTaskResult queryTaskResult(Id taskid) {
this.resultReadCount.incrementAndGet();
this.checkResultReadAllowed();
return super.queryTaskResult(taskid);
}
@Override
protected Iterator<HugeTaskResult> queryTaskResult(List<Id> taskIds) {
this.resultReadCount.incrementAndGet();
this.checkResultReadAllowed();
return super.queryTaskResult(taskIds);
}
@Override
protected void deleteTaskResultFromTx(Id taskId) {
if (this.failNextTaskResultDelete) {
this.failNextTaskResultDelete = false;
throw new HugeException("Mock task result delete failure");
}
super.deleteTaskResultFromTx(taskId);
}
@Override
protected boolean isLockedTask(String taskId) {
return false;
}
public <V> HugeTask<V> deleteFromDBForTest(Id id) {
return super.deleteFromDB(id);
}
public Iterator<HugeTask<Object>> queryMetadataTasksByStatus(
TaskStatus status) {
return super.queryTaskWithoutResultByStatus(status);
}
public boolean taskVertexExists(Id id) {
return this.vertexExists(id);
}
public boolean taskResultVertexExists(Id id) {
return this.vertexExists(HugeTaskResult.genId(id));
}
public TaskStatus taskStatus(Id id) {
return this.task(id, false).status();
}
public int resultReadCount() {
return this.resultReadCount.get();
}
public void resetResultReadCount() {
this.resultReadCount.set(0);
}
public void forbidResultRead(boolean forbid) {
this.forbidResultRead = forbid;
}
public void failNextTaskResultDelete() {
this.failNextTaskResultDelete = true;
}
public void allowTaskResultDelete() {
this.failNextTaskResultDelete = false;
}
public void closeAndShutdown() {
try {
this.close();
} finally {
this.schedulerExecutor.shutdownNow();
this.taskDbExecutor.shutdownNow();
this.schemaTaskExecutor.shutdownNow();
this.olapTaskExecutor.shutdownNow();
this.gremlinTaskExecutor.shutdownNow();
this.ephemeralTaskExecutor.shutdownNow();
this.serverInfoDbExecutor.shutdownNow();
}
}
private boolean vertexExists(Object id) {
return this.call(() -> {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
return vertex != null;
});
}
private void checkResultReadAllowed() {
if (this.forbidResultRead) {
throw new AssertionError("Unexpected task result read");
}
}
}
}