Add task name based tracing to scheduled tasks.

orientechnologies · Oct 31, 2022 · d533fc8 · d533fc8
1 parent 8c8e67c
commit d533fc8
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 14 deletions.
diff --git a/core/src/main/java/com/orientechnologies/orient/core/Orient.java b/core/src/main/java/com/orientechnologies/orient/core/Orient.java
@@ -411,6 +411,11 @@ public Orient shutdown() {
   }
 
   public TimerTask scheduleTask(final Runnable task, final long delay, final long period) {
+    return scheduleTask(task.getClass().getSimpleName(), task, delay, period);
+  }
+
+  public TimerTask scheduleTask(
+      final String taskName, final Runnable task, final long delay, final long period) {
     engineLock.readLock().lock();
     try {
       final TimerTask timerTask =
@@ -421,16 +426,10 @@ public void run() {
                 task.run();
               } catch (Exception e) {
                 OLogManager.instance()
-                    .error(
-                        this,
-                        "Error during execution of task " + task.getClass().getSimpleName(),
-                        e);
+                    .error(Orient.this, "Error during execution of task " + taskName, e);
               } catch (Error e) {
                 OLogManager.instance()
-                    .error(
-                        this,
-                        "Error during execution of task " + task.getClass().getSimpleName(),
-                        e);
+                    .error(Orient.this, "Error during execution of task " + taskName, e);
                 throw e;
               }
             }

diff --git a/...n/java/com/orientechnologies/orient/server/distributed/impl/ODistributedDatabaseImpl.java b/...n/java/com/orientechnologies/orient/server/distributed/impl/ODistributedDatabaseImpl.java
@@ -31,6 +31,7 @@
 import com.orientechnologies.common.profiler.OAbstractProfiler;
 import com.orientechnologies.common.profiler.OProfiler;
 import com.orientechnologies.common.thread.OThreadPoolExecutors;
+import com.orientechnologies.common.thread.TracingExecutorService;
 import com.orientechnologies.common.util.OCallable;
 import com.orientechnologies.orient.core.Orient;
 import com.orientechnologies.orient.core.command.OCommandDistributedReplicateRequest;
@@ -96,7 +97,7 @@ public class ODistributedDatabaseImpl implements ODistributedDatabase {
   private final AtomicLong operationsRunnig = new AtomicLong(0);
   private final ODistributedSynchronizedSequence sequenceManager;
   private final AtomicLong pending = new AtomicLong();
-  private final ThreadPoolExecutor requestExecutor;
+  private final TracingExecutorService requestExecutor;
   private final OLockManager lockManager = new OLockManagerImpl();
   private final Set<OTransactionId> inQueue = Collections.newSetFromMap(new ConcurrentHashMap<>());
   private OFreezeGuard freezeGuard;
@@ -173,7 +174,7 @@ public void endOperation() {
     this.localNodeName = manager.getLocalNodeName();
 
     this.requestExecutor =
-        (ThreadPoolExecutor)OThreadPoolExecutors.newScalingThreadPool(
+        OThreadPoolExecutors.newScalingThreadPool(
             String.format(
                 "OrientDB DistributedWorker node=%s db=%s", getLocalNodeName(), databaseName),
             0,
@@ -258,7 +259,7 @@ public Object getValue() {
             new OAbstractProfiler.OProfilerHookValue() {
               @Override
               public Object getValue() {
-                return (long) requestExecutor.getPoolSize();
+                return (long) ((ThreadPoolExecutor) requestExecutor).getPoolSize();
               }
             },
             "distributed.db.*.workerThreads");
@@ -305,6 +306,7 @@ public void reEnqueue(
     pending.incrementAndGet();
     Orient.instance()
         .scheduleTask(
+            String.format("DistributedDatabase[%s].reEnqueue", databaseName),
             () -> {
               try {
                 processRequest(
@@ -343,6 +345,9 @@ public void processRequest(
                 + "' discarding");
       }
     }
+    final String taskName =
+        String.format(
+            "DistributedDatabase[%s].processRequest.%s", getDatabaseName(), task.getName());
     synchronized (this) {
       task.received(request, this);
       manager.messageReceived(request);
@@ -364,7 +369,7 @@ public void processRequest(
                     }
                   };
               try {
-                this.requestExecutor.submit(executeTask);
+                this.requestExecutor.submit(taskName, executeTask);
               } catch (RejectedExecutionException e) {
                 task.finished(this);
                 this.lockManager.unlock(guards);
@@ -380,6 +385,7 @@ public void processRequest(
       } else {
         try {
           this.requestExecutor.submit(
+              taskName,
               () -> {
                 execute(request);
               });
@@ -892,7 +898,7 @@ public long getReceivedRequests() {
 
   @Override
   public long getProcessedRequests() {
-    return requestExecutor.getCompletedTaskCount();
+    return ((ThreadPoolExecutor) requestExecutor).getCompletedTaskCount();
   }
 
   public void onDropShutdown() {
@@ -1310,7 +1316,7 @@ public String dump() {
         "\n\nDATABASE '" + databaseName + "' ON SERVER '" + manager.getLocalNodeName() + "'");
 
     buffer.append("\n- MESSAGES IN QUEUES");
-    buffer.append(" (" + (requestExecutor.getPoolSize()) + " WORKERS):");
+    buffer.append(" (" + (((ThreadPoolExecutor) requestExecutor).getPoolSize()) + " WORKERS):");
 
     return buffer.toString();
   }