When getting the result of the generated objects, a vector with one element returns "RNumeric", while vector with many elements returns "RNumericVector". I believe both should return "RNumericVector".

In the R script I am type casting to vector:

newscale <- as.vector(results$par)

When I get the generated objects, I am having to check the type before I place the objects into a List.

RTaskToken rTaskToken = rBroker.submit(rTask);
RTaskResult rTaskResult = rTaskToken.getResult();
foreach (var rslt in rTaskResult.getGeneratedObjects())
{
    if (rslt.Name.ToLower().Equals("newscale"))
    {
        var rsltList = new List<double?>();
        if (rslt.RClass == "numeric" && rslt.Type == "primitive")
        {
            rsltList.Add((double?)rslt.Value);
        }
        else
        {
            rsltList = ((IEnumerable<double?>)rslt.Value).ToList();
        }
    }
}

When I'm doing many concurrent pooled task execution using single RBroker, sometimes RBrokerWorkerManager method fails with an exception:

Exception: Exception thrown: 'System.Exception' in DeployRBroker8.0.5.dll ("RBrokerEngine: brokerEngineExecutor.run ex=System.Exception: RBrokerEngine: processing task DeployRBroker.PooledTask, ex=System.Exception: RBrokerEngine: taskTokenListener callback not found for DeployRBroker.PooledTask, unexpected error. at DeployRBroker.RBrokerEngine.RBrokerWorkerManager() at DeployRBroker.RBrokerEngine.RBrokerWorkerManager()"). Exception thrown: 'System.Exception' in DeployRBroker8.0.5.dll ("RBrokerEngine: brokerEngineExecutor.run ex=System.Exception: RBrokerEngine: processing task DeployRBroker.PooledTask, ex=System.Exception: RBrokerEngine: taskTokenListener callback not found for DeployRBroker.PooledTask, unexpected error. at DeployRBroker.RBrokerEngine.RBrokerWorkerManager() at DeployRBroker.RBrokerEngine.RBrokerWorkerManager()")

Once the RBrokerWorkerManager method throws the exception, RBroker basically stops task processing and new tasks keep queuing.
I tested on series of 100 concurrent calls with parallelTaskLimit = 6.

It happens sometimes in case when a task has been sent to submit and was successfully added into pending tasks queue (m_pendingHighPriorityQueue/m_pendingLowPriorityQueue). At the same time the task could be picked up in different thread by RBrokerWorkerManager, although the task was not completely submitted yet. In some cases RBrokerWorkerManager tries to take the task token listener from m_taskTokenListenerMapbefore it was created and added there in submit method. So it cause that m_taskTokenListenerMap.TryRemove returns null since such key was not found. Then RBrokerWorkerManager throws an exception, which I described above.
I haven't found any logic that would restart RBrokerWorkerManager to continue task processing after failure.

Assuming that I see two point to improve:

1. RBrokerWorkerManager should restart after fail
2. A task and its result token object should be added to the m_taskTokenListenerMap dictionary before the task would be queued to process.

There are important files that Microsoft projects should all have that are not present in this repository. A pull request has been opened to add the missing file(s). When the pr is merged this issue will be closed automatically.

Microsoft teams can learn more about this effort and share feedback within the open source guidance available internally.

Hi! It's me again.

As you know the RBrokerListenerManagermethod, which handles task result, iterates m_liveTaskTokensqueue by task token. If current task is done, it invokes onTaskCompletedfor this task and removes a task from head of the m_liveTaskTokens queue. This logic works fine while all tasks are done or were executed in single thread or one by one. But once one of tasks is not done, this task continue to be the first to out and will be removed from the queue with next completed task. Consequently some of task would not be notified as completed and some would be handled twice or more times.
So using for this case a queue is definitely wrong approach. Please replace ConcurrentBag<RTaskToken>with a ConcurrentDictionary<RTaskToken, RTask>, which supports direct remove a particular instance.

Cheers!