FWD » Core Development » Base Language

There are some failing tests. Is it the deployment or it's fine?

This is the server log of errors.

There is one point in AppServerConnectionPool that is bothering me. It's used only by LegacyJavaAppserverClient and LegacyServiceHandler:

   /** The pool of workers dedicated to executing legacy services. */
   private PriorityBlockingQueue<LegacyServiceWorker> workers = new PriorityBlockingQueue<>();

   /** Mapping of workers by their connection ID. */
   private Map<String, LegacyServiceWorker> workersByConnection = new ConcurrentHashMap<>();

   public boolean dispatch(String connectionID, Consumer<LegacyServiceWorker> work, String target)
   {
      // TODO: increase the workers based on the appserver settings (max_agents, etc)????

      // find a worker
      LegacyServiceWorker worker;
      if (connectionID == null)
      {
         try
         {
            worker = workers.take();
         }
         catch (InterruptedException e1)
         {
            // work interrupted
            LOG.log(Level.SEVERE, "Interrupted worker for " + target);

            return false;
         }
      }
      else
      {
         worker = workersByConnection.get(connectionID);

         if (worker == null)
         {
            // work interrupted
            LOG.log(Level.SEVERE, 
                    "Could not find worker for connection " + connectionID + " when processing " + target);

            return false;
         }
      }

      Throwable[] exc = new Throwable[1];

      CountDownLatch down = new CountDownLatch(1);
      worker.addWork(() ->
      {
         try
         {
            // this ensures the queue remains sorted...
            workers.remove(worker);
            workers.add(worker);

            work.accept(worker);
         }
         catch (Throwable t)
         {
            exc[0] = t;
         }
         finally
         {
            down.countDown();
         }
      });

      boolean done = false;
      try
      {
         if (timeout == 0)
         {
            down.await();
            done = true;
         }
         else
         {
            done = down.await(timeout, TimeUnit.SECONDS);
         }
      }
      catch (InterruptedException e)
      {
         // work interrupted
         LOG.log(Level.SEVERE, "Interrupted worker for " + target);

         return false;
      }
      finally
      {
         // this ensures the queue remains sorted...
         workers.remove(worker);
         workers.add(worker);
      }

      if (exc[0] != null)
      {
         if (exc[0] instanceof RuntimeException)
         {
            throw (RuntimeException) exc[0];
         }

         throw new RuntimeException(exc[0]);
      }

      if (!done)
      {
         LOG.log(Level.SEVERE, type + " call " + target + " terminated, timeout elapsed.");
         return false;
      }

      return true;
   }

I think the queue workers doesn't stay sorted as expected. The calls to dispatch sometimes have connectionId and sometimes not, so sometimes the if block is executed, other times the else. When there is a connectionID the taken worker is not the next in the queue, but the logic below still applies.

addWork adds a task. When the task starts execution the worker it uses get's removed from its place in the queue (even if it's not the next element) and added back as last. The same operation is repeated on task execution completed (in the finally block). This method is not synchronized, also tasks will be executed by different workers in different threads, so remove-add is not one atomic operation.

workers.take() removes a worker, but adds it back right before the execution of the task starts.

I would suggest to remove the remove/add from both places:

            workers.remove(worker);
            workers.add(worker);