Understanding queues and nodes #22

wlandau · 2023-02-27T01:02:44Z

wlandau
Feb 27, 2023

As discussed in #18, I would like to use https://github.com/shikokuchuo/mirai#example-1-connecting-to-remote-servers--remote-server-queues to distribute tasks across jobs on a cluster (and eventually, across AWS Batch jobs on the same local network). Each cluster or AWS Batch job will run one R process that calls mirai::server() which connects back to the client, and I assume each server process will accept one task at a time. Tasks will vary in runtime, and there will be more tasks than servers/nodes, so it sounds like an active queue will be useful.

I have learned a lot from the README, the documentation of daemons() and server(), my experiments with the code, and your helpful comments in #18. However, I still do not understand how queueing or nodes work in https://github.com/shikokuchuo/mirai#example-1-connecting-to-remote-servers--remote-server-queues. I would appreciate it if we could revisit the topic at a beginner level. My code examples below use commit 812e93f. I apologize if my questions repeat things I already asked in #19.

What I think I understand so far

Aside from autoscaling and crash detection, here is how I see my use case playing out. The following example has 2 workers and only a few tasks, but I envision scaling up to hundreds of each.

First, I open a TCP socket on the client:

# from xx.xx.xx.175
daemons("tcp://xx.xx.xx.175:5555")

and then on a different computer on the network I run a server process

# from xx.xx.xx.138
server("tcp://xx.xx.xx.175:5555")

and on a third computer I run a third server process

# from xx.xx.xx.176
server("tcp://xx.xx.xx.175:5555")

Then the client can send jobs for the two servers to run. Below, I experiment with a task load of 6 jobs for 2 servers.

# from xx.xx.xx.175
tasks <- list(
  mirai({Sys.sleep(5); paste0("task1_server", gsub("^.*\\.", "", getip::getip()))}),
  mirai({Sys.sleep(3); paste0("task2_server", gsub("^.*\\.", "", getip::getip()))}),
  mirai({Sys.sleep(4); paste0("task3_server", gsub("^.*\\.", "", getip::getip()))}),
  mirai({Sys.sleep(2); paste0("task4_server", gsub("^.*\\.", "", getip::getip()))}),
  mirai({Sys.sleep(8); paste0("task5_server", gsub("^.*\\.", "", getip::getip()))}),
  mirai({Sys.sleep(7); paste0("task6_server", gsub("^.*\\.", "", getip::getip()))})
)
for (i in seq_len(60)) {
  Sys.sleep(1)
  seconds <- paste0(i, "s")
  print(paste(c(seconds, purrr::map_chr(tasks, ~.x$data)), collapse = " "))
}

The output I see from the loop on the client is:

[1] "1s NA NA NA NA NA NA"
[1] "2s NA NA NA NA NA NA"
[1] "3s NA task2_server138 NA NA NA NA"
[1] "4s NA task2_server138 NA NA NA NA"
[1] "5s task1_server176 task2_server138 NA task4_server138 NA NA"
[1] "6s task1_server176 task2_server138 NA task4_server138 NA NA"
[1] "7s task1_server176 task2_server138 NA task4_server138 NA NA"
[1] "8s task1_server176 task2_server138 NA task4_server138 NA NA"
[1] "9s task1_server176 task2_server138 task3_server176 task4_server138 NA NA"
[1] "10s task1_server176 task2_server138 task3_server176 task4_server138 NA NA"
[1] "11s task1_server176 task2_server138 task3_server176 task4_server138 NA NA"
[1] "12s task1_server176 task2_server138 task3_server176 task4_server138 NA task6_server138"
[1] "13s task1_server176 task2_server138 task3_server176 task4_server138 NA task6_server138"
[1] "14s task1_server176 task2_server138 task3_server176 task4_server138 NA task6_server138"
[1] "15s task1_server176 task2_server138 task3_server176 task4_server138 NA task6_server138"
[1] "16s task1_server176 task2_server138 task3_server176 task4_server138 NA task6_server138"
[1] "17s task1_server176 task2_server138 task3_server176 task4_server138 task5_server176 task6_server138"

From this output, we can draw a picture of which tasks are running at which times. In the plot below, each row is a task. Time advances from left to right, and each task is shaded according to the server running it at a given time point.

I really like the way mirai scheduled these tasks. Even though there were more tasks than servers, and even though the tasks varied in length, each worker was as busy as possible, and tasks on the queue began as soon as possible. Unless there is something special about auto-scaling (e.g. #19 and upscaling) or the integer-length runtimes I coded, the scheduling in this part of mirai is already perfect. In my situation, I find it hard to see why I would need an active queue or the "nodes" functionality. (Admittedly, I do not have grasp of the mirai-specific versions of these concepts.) Am I missing something relevant to my scenario?

Nodes and queues?

In https://github.com/shikokuchuo/mirai#example-1-connecting-to-remote-servers--remote-server-queues, you explain that server() accepts a nodes argument. That section and https://github.com/shikokuchuo/mirai#example-2-connecting-to-remote-servers-through-a-local-server-queue describe active queues. So my questions are:

What exactly does the nodes argument do?
What exactly is an active queue in mirai, and how do it work? (I.e. where does it run and what does it orchestrate?)
Given the positive outcome from my code example above, do I even need (1) or (2)?

I am sorry if I am missing something obvious from the code usage examples, documentation, or #18, but would you be willing to explain it to me at a more basic level what these features are and the specific scenarios that motivate them?

Answered by shikokuchuo

Feb 27, 2023

Let me answer (3) first – if you don’t need an active queue, don’t use it. The underlying NNG logic and implementation is very robust and I have recommended throughout the documentation that if this is suitable then it should be used.

The problem we have is that when we send the tasks we have no way of knowing the task length a priori. In your example, the tasks are still roughly the same length so the solution is more or less acceptable.

Let me give you an extreme counter-example: odd number tasks length 1, even number tasks length 10. As NNG round-robins*, the odd number tasks are all sent to server 1, and the even ones to server 2. Server 1 will be idle after 3 seconds, However the tot…

View full answer

shikokuchuo · 2023-02-27T11:53:29Z

shikokuchuo
Feb 27, 2023
Maintainer

Let me answer (3) first – if you don’t need an active queue, don’t use it. The underlying NNG logic and implementation is very robust and I have recommended throughout the documentation that if this is suitable then it should be used.

The problem we have is that when we send the tasks we have no way of knowing the task length a priori. In your example, the tasks are still roughly the same length so the solution is more or less acceptable.

Let me give you an extreme counter-example: odd number tasks length 1, even number tasks length 10. As NNG round-robins*, the odd number tasks are all sent to server 1, and the even ones to server 2. Server 1 will be idle after 3 seconds, However the total time taken will be 30 seconds.

This is the reason for an active queue.

*NNG is not so dumb it also responds to back-pressure from the socket, but only in the case where messages are sent faster than socket buffers can be cleared. Buffers are tuneable at the NNG level but there are also system level TCP socket buffers - in short this is not something we can control reliably.

1 reply

wlandau Feb 27, 2023
Author

Let me give you an extreme counter-example: odd number tasks length 1, even number tasks length 10. As NNG round-robins*, the odd number tasks are all sent to server 1, and the even ones to server 2. Server 1 will be idle after 3 seconds, However the total time taken will be 30 seconds.

Ah, I see now. Thanks for explaining. It sounds like an active queue would be useful after all, and you have given me a testable example where it makes a difference.

This is the reason for an active queue.

Yes, I agree now. Thank you for patiently explaining it to me.

shikokuchuo · 2023-02-27T12:05:50Z

shikokuchuo
Feb 27, 2023
Maintainer

(1) Specifying nodes means using an active server queue. It will launch on the local machine a background process running this branch of the server() code: https://github.com/shikokuchuo/mirai/blob/main/R/mirai.R#L84

2 replies

wlandau Feb 27, 2023
Author

In that case, what does nodes mean in terms of how the active server queue works? Say I have a pipeline that auto-scales between 4 and 8 servers, depending on the workload. What value should I set nodes to be in the call to daemons()?

shikokuchuo Feb 27, 2023
Maintainer

In a new commit 51adb17 v0.7.2.9007 - this is now possible. You can set nodes = 8 and auto-scale your servers however you want.

shikokuchuo · 2023-02-27T12:19:17Z

shikokuchuo
Feb 27, 2023
Maintainer

(2) What the active queue does is act as a relay or switch - it sits in the middle and forwards tasks between the client and an end server.

The logic it has is to only forward tasks to servers that are free (idle). This is the key difference! As we cannot know a priori the task length, we should not allocate them all at once - they need to be queued. Now we just need to poll for servers becoming free and send tasks to the server if there are remaining tasks in the queue. The polling is why we run this in a background process.

So if you now call:

daemons("tcp://:5555", n = 2)

(I find it easier to specify all interfaces on the client, also partial matching works so 'n' is as good as 'nodes')

and

# from xx.xx.xx.138
server("tcp://xx.xx.xx.175:5555")

# from xx.xx.xx.176
server("tcp://xx.xx.xx.175:5556")

[]* Note the increment of the port number

You will see now that waiting tasks will be allocated to a server as soon as it becomes available.

22 replies

HenrikBengtsson May 2, 2023

Thank you.

"online" means there is an active pipe connection established (the server is fully ready to receive tasks). This is a boolean.
...
There is meant to be a one to one relationship between a URL and server instance when using dispatcher. Connecting more than one server to a unique URI leads to unsupported behaviour. You will notice that tasks do not get done if the 'online' status is zero.

Does this mean that you can have at most nrow(daemons()$daemons) servers connection back ("signing in")?

However, I'm still confused. Trying your example, I get:

> library(mirai)
> daemons(1, maxtasks = 1L)
[1] 1

> daemons()
$connections
[1] 1

$daemons
                                                    online instance assigned
abstract://682cee060c4fab023a5c9b1248b479dc621afa17      1        1        0
                                                    complete
abstract://682cee060c4fab023a5c9b1248b479dc621afa17        0

From the above I conclude there is one "server" connected (instance = 1) and there is a "pipe" ready to it (online = 1).

Continuing:

> m <- mirai(1)
> daemons()
$connections
[1] 1

$daemons
                                                    online instance assigned
abstract://682cee060c4fab023a5c9b1248b479dc621afa17      0        1        1
                                                    complete
abstract://682cee060c4fab023a5c9b1248b479dc621afa17        1

From this, plus the previous state, I conclude that the task was assigned (assigned = 1) and completed (complete = 1). [BTW, should it be called completed instead of complete in parallel with assigned?]

From instance = 1, I conclude the server is still connected, but I don't understand why online = 0 and not online = 1? What changed? Because, then sum(daemons()$daemons[, "online"]) says there are zero workers available.

Collecting the results makes no difference;

> m$data
[1] 1
> daemons()
$connections
[1] 1

$daemons
                                                    online instance assigned
abstract://38455164d96b891aea32836a026321e3c696ff87      0        1        1
                                                    complete
abstract://38455164d96b891aea32836a026321e3c696ff87        1

Continuing with your example, I call and get:

> launch_server(row.names(daemons()$daemons)[[1L]], maxtasks = 1L)
> daemons()
$connections
[1] 1

$daemons
                                                    online instance assigned
abstract://682cee060c4fab023a5c9b1248b479dc621afa17      1        2        0
                                                    complete
abstract://682cee060c4fab023a5c9b1248b479dc621afa17        0

So, I'm still confused, because above you said: "Connecting more than one server to a unique URI leads to unsupported behaviour.". Are there one or two servers running/connected at this point? From instance = 2, I'd say two, but that contradicts it's only possible to have at most one.

Continuing on:

> m <- mirai(2)
> daemons()
$connections
[1] 1

$daemons
                                                    online instance assigned
abstract://682cee060c4fab023a5c9b1248b479dc621afa17      0        2        1
                                                    complete
abstract://682cee060c4fab023a5c9b1248b479dc621afa17        1

Hmm... online went back to 0 again.

Just to make sure, I got the correct results back:

> m$data
[1] 2

That's good, but if I try to launch yet another task, it does not get resolved;

> m <- mirai(3)
> m$data
'unresolved' logi NA

It stays like this forever. At this point I see:

> daemons()
$connections
[1] 1

$daemons
                                                    online instance assigned
abstract://9667cb6e2930a6392cdd49f5b4e885104f9cfaf4      0        2        1
                                                    complete
abstract://9667cb6e2930a6392cdd49f5b4e885104f9cfaf4        1

I'm using mirai 0.8.3 and nanonext 0.8.2 here.

shikokuchuo May 2, 2023
Maintainer

Yes, what you are experiencing is intended. The design was primarily to facilitate the scaling algorithms in crew.

When a server disconnects, online goes from 1 to 0. The instance and task stats remain the same.
When a new server connects at the same URL, online goes from 0 to 1, instance increases by 1, and the task stats are reset to 0.

Perhaps it is not as obvious to a new user - I am happy to add to the documentation at least to make it clearer.

Does this mean that you can have at most nrow(daemons()$daemons) servers connection back ("signing in")?

Yes, the number of daemons 'n' can be thought of as the maximum servers that can be connected at any one time. crew is able to scale up or down from 0 to 'n'.

So, I'm still confused, because above you said: "Connecting more than one server to a unique URI leads to unsupported behaviour.". Are there one or two servers running/connected at this point? From instance = 2, I'd say two, but that contradicts it's only possible to have at most one.

At the point you mention, one server is connected. It is the 2nd instance of the server at that URL. That is how 'online' and 'instance' are meant to be read.

It stays like this forever. At this point I see:

At this point, the server has tasked out and disconnected again, hence 'online' is 0, and your mirai remain unresolved.

I hope the above helps.

HenrikBengtsson May 2, 2023

Thank you.

Perhaps it is not as obvious to a new user - I am happy to add to the documentation at least to make it clearer.

Yes, I think that would be helpful. Right now I'm not sure if I'm overthinking it, but after trying to understand what daemons()$daemons represents, I tried to reverse engineer it from different trial-and-error examples, and I think I ended up in a rabbit hole doing so.

Regarding online: So, if I ever see it go from 1 to 0, does that indicate something went wrong, e.g. the server terminated, or the network connection went down? Because as soon as I've got a server connected, and I don't shut it down intentionally, or it times out, it should stay at online = 1, correct?

Is it correct that:

online <- (daemons()$daemons[, "online"] == 1L)

is relevant for knowing the current state of the "cluster", whereas:

stats <- mirai::daemons()$daemons[, c("instance", "assigned", "complete")]

is historical data for what happened in the past (and irrelevant for task/worker orchestration)?

If you want to scale up and down the servers, is it correct that it's only the online flag you should look at, e.g. only launch servers for daemons with online = 0?

shikokuchuo May 2, 2023
Maintainer

Regarding online: So, if I ever see it go from 1 to 0, does that indicate something went wrong, e.g. the server terminated, or the network connection went down? Because as soon as I've got a server connected, and I don't shut it down intentionally, or it times out, it should stay at online = 1, correct?

Correct, in normal course, it should never go from 1 to 0 by itself unless it has been set to time out etc. However in case it does, i.e. it has really crashed or network has gone down, you can simply replace it with another 'instance'.

Is it correct that:
online <- (daemons()$daemons[, "online"] == 1L)
is relevant for knowing the current state of the "cluster", whereas:
stats <- mirai::daemons()$daemons[, c("instance", "assigned", "complete")]
is historical data for what happened in the past (and irrelevant for task/worker orchestration)?

If you want to scale up and down the servers, is it correct that it's only the online flag you should look at, e.g. only launch servers for daemons with online = 0?

Absolutely - only 'online' is relevant for the current status. Only launch new server 'instances' when online = 0. crew uses the historical data to determine when to scale up or down e.g. if some servers have hardly been used then not re-launch those when they time out.

Thank you for going over this - it is helpful to have a fresh perspective as we have been developing it intensively for so long - what is obvious to us may well not be as intuitive.

shikokuchuo May 3, 2023
Maintainer

I found the relevant description of the status matrix in the readme - I have transferred this to the daemons() function documentation in 31e0413 and added some clarifications as part of 0d4e586.

I agree completely with Henrik's ideas for improving the readability of the stats, however it is currently a trade-off with efficiency as per Will's comment. Although we may know how to interact with data frames in a performant manner, we cannot assume this for all users, hence I prefer to maintain the matrix, which is efficient by default.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Understanding queues and nodes #22

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Replies: 3 comments 25 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Select a reply

Understanding queues and nodes #22

wlandau Feb 27, 2023

What I think I understand so far

Nodes and queues?

Replies: 3 comments · 25 replies

shikokuchuo Feb 27, 2023 Maintainer

wlandau Feb 27, 2023 Author

shikokuchuo Feb 27, 2023 Maintainer

wlandau Feb 27, 2023 Author

shikokuchuo Feb 27, 2023 Maintainer

shikokuchuo Feb 27, 2023 Maintainer

HenrikBengtsson May 2, 2023

shikokuchuo May 2, 2023 Maintainer

HenrikBengtsson May 2, 2023

shikokuchuo May 2, 2023 Maintainer

shikokuchuo May 3, 2023 Maintainer

wlandau
Feb 27, 2023

Replies: 3 comments 25 replies

shikokuchuo
Feb 27, 2023
Maintainer

wlandau Feb 27, 2023
Author

shikokuchuo
Feb 27, 2023
Maintainer

wlandau Feb 27, 2023
Author

shikokuchuo Feb 27, 2023
Maintainer

shikokuchuo
Feb 27, 2023
Maintainer

shikokuchuo May 2, 2023
Maintainer

shikokuchuo May 2, 2023
Maintainer

shikokuchuo May 3, 2023
Maintainer