Broker (OpenTripPlanner 1.3.0 API)

java.lang.Object
- org.opentripplanner.analyst.broker.Broker

All Implemented Interfaces:

Runnable
```
public class Broker
extends Object
implements Runnable
```
This class tracks incoming requests from workers to consume Analyst tasks, and attempts to match those requests to enqueued tasks. It aims to draw tasks fairly from all users, and fairly from all jobs within each user, while attempting to respect the graph affinity of each worker (give it tasks that require the same graph it has been working on recently). When no work is available or no workers are available, the polling functions return immediately, avoiding spin-wait. When they are receiving no work, workers are expected to disconnect and re-poll occasionally, on the order of 30 seconds. This serves as a signal to the broker that they are still alive and waiting. TODO if there is a backlog of work (the usual case when jobs are lined up) workers will constantly change graphs. Because (at least currently) two users never share the same graph, we can get by with pulling tasks cyclically or randomly from all the jobs, and just actively shaping the number of workers with affinity for each graph by forcing some of them to accept tasks on graphs other than the one they have declared affinity for. This could be thought of as "affinity homeostasis". We will constantly keep track of the ideal proportion of workers by graph (based on active queues), and the true proportion of consumers by graph (based on incoming requests) then we can decide when a worker's graph affinity should be ignored and what it should be forced to. It may also be helpful to mark jobs every time they are skipped in the LRU queue. Each time a job is serviced, it is taken out of the queue and put at its end. Jobs that have not been serviced float to the top.

Nested Class Summary

Nested Classes
Modifier and Type	Class and Description
`static class`	`Broker.WrappedResponse` We wrap responses in a class that has a machine ID, and then put them in a TreeSet so that the machine with the lowest ID on a given graph always gets single-point work.

Field Summary

Fields
Modifier and Type	Field and Description
`CircularList<Job>`	`jobs`
`int`	`MAX_TASKS_PER_WORKER` the most tasks to deliver to a worker at a time
`static long`	`WORKER_STARTUP_TIME` How long to give workers to start up (in ms) before assuming that they have started (and starting more on a given graph if they haven't.

Constructor Summary

Constructors
Constructor and Description

Broker(Properties brokerConfig, String addr, int port)

Constructors
Constructor and Description
`Broker(Properties brokerConfig, String addr, int port)`

Method Summary

All Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`boolean`	`anyJobsActive()`
`void`	`createWorkersForGraph(String graphId)` Create workers for a given job, if need be
`boolean`	`deleteJob(String jobId)` delete a job
`org.glassfish.grizzly.http.server.Response`	`deletePriorityTask(int taskId)` Marks the specified priority request as completed, and returns the suspended Response object for the connection that submitted the priority request (the UI), which probably still waiting to receive a result back over the same connection.
`boolean`	`deliver(Job job, org.glassfish.grizzly.http.server.Response response)` Attempt to hand some tasks from the given job to a waiting consumer connection.
`void`	`deliverHighPriorityTasks(String graphId)` attempt to deliver high priority tasks via side channels, or move them into normal channels if need be
`void`	`deliverTasks()` This method checks whether there are any high-priority tasks or normal job tasks and attempts to match them with waiting workers.
`void`	`enqueuePriorityTask(AnalystClusterRequest task, org.glassfish.grizzly.http.server.Response response)` Enqueue a task for execution ASAP, planning to return the response over the same HTTP connection.
`void`	`enqueueTasks(List<AnalystClusterRequest> tasks)` Enqueue some tasks for queued execution possibly much later.
`Job`	`findJob(AnalystClusterRequest task)` find the job for a task, creating it if it does not exist
`Job`	`findJob(String jobId)` find the job for a jobId, or null if it does not exist
`Job`	`getJobForTask(int taskId)` This uses a linear search through jobs, which should not be problematic unless there are thousands of simultaneous jobs.
`boolean`	`markTaskCompleted(int taskId)` Take a normal (non-priority) task out of a job queue, marking it as completed so it will not be re-delivered.
`void`	`registerSinglePointChannel(String graphAffinity, Broker.WrappedResponse response)` Register an HTTP connection that can be used to send single point requests directly to workers, bypassing normal task distribution channels.
`void`	`registerSuspendedResponse(String graphId, org.glassfish.grizzly.http.server.Response response)` Consumer long-poll operations are enqueued here.
`boolean`	`removeSinglePointChannel(String graphAffinity, Broker.WrappedResponse response)` Remove a single point channel because the connection was closed.
`boolean`	`removeSuspendedResponse(String graphId, org.glassfish.grizzly.http.server.Response response)` When we notice that a long poll connection has closed, we remove it here.
`void`	`run()`
`boolean`	`workersAvailableForGraph(String graphId)`

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - jobs
```
public final CircularList<Job> jobs
```
  - MAX_TASKS_PER_WORKER
```
public final int MAX_TASKS_PER_WORKER
```
    the most tasks to deliver to a worker at a time
    
    See Also:
    
    Constant Field Values
  - WORKER_STARTUP_TIME
```
public static final long WORKER_STARTUP_TIME
```
    How long to give workers to start up (in ms) before assuming that they have started (and starting more on a given graph if they haven't.
    
    See Also:
    
    Constant Field Values
- Constructor Detail
  - Broker
```
public Broker(Properties brokerConfig,
              String addr,
              int port)
```
- Method Detail
  - enqueuePriorityTask
```
public void enqueuePriorityTask(AnalystClusterRequest task,
                                org.glassfish.grizzly.http.server.Response response)
```
    Enqueue a task for execution ASAP, planning to return the response over the same HTTP connection. Low-reliability, no re-delivery.
  - deliverHighPriorityTasks
```
public void deliverHighPriorityTasks(String graphId)
```
    attempt to deliver high priority tasks via side channels, or move them into normal channels if need be
  - enqueueTasks
```
public void enqueueTasks(List<AnalystClusterRequest> tasks)
```
    Enqueue some tasks for queued execution possibly much later. Results will be saved to S3.
  - workersAvailableForGraph
```
public boolean workersAvailableForGraph(String graphId)
```
  - createWorkersForGraph
```
public void createWorkersForGraph(String graphId)
```
    Create workers for a given job, if need be
  - registerSuspendedResponse
```
public void registerSuspendedResponse(String graphId,
                                      org.glassfish.grizzly.http.server.Response response)
```
    Consumer long-poll operations are enqueued here.
  - removeSuspendedResponse
```
public boolean removeSuspendedResponse(String graphId,
                                       org.glassfish.grizzly.http.server.Response response)
```
    When we notice that a long poll connection has closed, we remove it here.
  - registerSinglePointChannel
```
public void registerSinglePointChannel(String graphAffinity,
                                       Broker.WrappedResponse response)
```
    Register an HTTP connection that can be used to send single point requests directly to workers, bypassing normal task distribution channels.
  - removeSinglePointChannel
```
public boolean removeSinglePointChannel(String graphAffinity,
                                        Broker.WrappedResponse response)
```
    Remove a single point channel because the connection was closed.
  - deliverTasks
```
public void deliverTasks()
                  throws InterruptedException
```
    This method checks whether there are any high-priority tasks or normal job tasks and attempts to match them with waiting workers. It blocks until there are tasks or workers available.
    
    Throws:
    
    InterruptedException
  - getJobForTask
```
public Job getJobForTask(int taskId)
```
    This uses a linear search through jobs, which should not be problematic unless there are thousands of simultaneous jobs.
    
    Returns:
    
    a Job object that contains the given task ID.
  - deliver
```
public boolean deliver(Job job,
                       org.glassfish.grizzly.http.server.Response response)
```
    Attempt to hand some tasks from the given job to a waiting consumer connection. The write will fail if the consumer has closed the connection but it hasn't been removed from the connection queue yet. This can happen because the Broker methods are synchronized, and the removal action may be waiting to get the monitor while we are trying to distribute tasks here.
    
    Returns:
    
    whether the handoff succeeded.
  - markTaskCompleted
```
public boolean markTaskCompleted(int taskId)
```
    Take a normal (non-priority) task out of a job queue, marking it as completed so it will not be re-delivered. TODO maybe use unique delivery receipts instead of task IDs to handle redelivered tasks independently
    
    Returns:
    
    whether the task was found and removed.
  - deletePriorityTask
```
public org.glassfish.grizzly.http.server.Response deletePriorityTask(int taskId)
```
    Marks the specified priority request as completed, and returns the suspended Response object for the connection that submitted the priority request (the UI), which probably still waiting to receive a result back over the same connection. A HttpHandler thread can then pump data from the DELETE body back to the origin of the request, without blocking the broker thread. TODO rename to "deregisterSuspendedProducer" and "deregisterSuspendedConsumer" ?
  - run
```
public void run()
```
    Specified by:
    
    run in interface Runnable
  - findJob
```
public Job findJob(AnalystClusterRequest task)
```
    find the job for a task, creating it if it does not exist
  - findJob
```
public Job findJob(String jobId)
```
    find the job for a jobId, or null if it does not exist
  - deleteJob
```
public boolean deleteJob(String jobId)
```
    delete a job
  - anyJobsActive
```
public boolean anyJobsActive()
```

Class Broker

Nested Class Summary

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

jobs

MAX_TASKS_PER_WORKER

WORKER_STARTUP_TIME

Constructor Detail

Broker

Method Detail

enqueuePriorityTask

deliverHighPriorityTasks

enqueueTasks

workersAvailableForGraph

createWorkersForGraph

registerSuspendedResponse

removeSuspendedResponse

registerSinglePointChannel

removeSinglePointChannel

deliverTasks

getJobForTask

deliver

markTaskCompleted

deletePriorityTask

run

findJob

findJob

deleteJob

anyJobsActive