post-job-dispatcher.md

Fullstack Speedrun: browser based job dispatcher

I love a good challenge! This is why I become fullstack developer, we never get the simple tasks 😠, there's always a catch: sometime there is technology/hardware limitation, sometime security demands, or more than often plain old tight deadlines.

This post is based on such a solution, as part of my job I needed to devise a quick POC for a job dispatcher where there were some technology limitation: the clients could connect to the server only within a browser embedded within another external application, the browser in question was of course IE 11 and we had no control on the external application. The only resource given to the task was my time, and not much of it...

The solution I found is based on the well known socket.io, it is mature and robust server-client socket communication library, it was quick to setup and had no issues with the limitation I had. I used an existing express server and extend it to reduce dev time, end to end it took me around 100 lines of code and around half of day to implement and test (excluding the actual processing task).

Server

State

This is the server state, the data model where I kept the jobs and workers data.

const initState = () => ({
  sockets: [],
  sIndex: 0,
  jobs: [],
  pending: [],
  results: []
})

• sockets - connected workers/clients.
• sIndex - current worker index (round-robin).
• jobs - jobs need to be processed.
• pending - jobs that are currently being processed.
• results - the processed results returned from the workers.

Listeners

We need to initiate the socket on every connection, I kept it simple and left up error events and handling.

const server = require('http').createServer(app) // app - connect server (express)
const io = require('socket.io')(server)
const state = initState()
io.on('connection', socket => {
    socket.on('register', data => {
      if (data && data.type === 'my-worker-type') state.sockets.push(socket)
    })
    socket.on('disconnect', () => {
      const index = state.sockets.findIndex(item => socket.id === item.id)
      if (index > -1) {
        state.sockets.splice(index, 1)
      }
    })
    socket.on('result', result => {
      const index = state.pending.findIndex(el => el.socket.id === socket.id)
      if (index > -1) state.pending.splice(index, 1)
      state.results.push(result)
    })
  })

• connection - top level listener fired when a client (socket) is connected for the first time.
• disconnect - socket listener, this event is fired when a client is disconnecting from the server. We search and removed the client from the sockets array.
• register - socket listener, this is a custom event that the client needs to fire in order to register itself as a worker (to be added to the sockets array).
• result - socket listener, this is a custom event that a worker fires with the process result.

The dispatch loop

const dispatchJob = () => {
  state.sIndex++
  // No jobs, no workers
  if (state.jobs.length === 0 || state.sockets.length === 0) return
  // No out of bound index
  if (state.sIndex > state.sockets.length) state.sIndex = 0
  // No worker (should not happen)
  if (!state.sockets[state.sIndex]) return
  // Getting worker socket
  const socket = state.sockets[state.sIndex]
  // If socket on pending list -> worker is busy
  if (state.pending.findIndex(el => el.socket.id === socket.id) !== -1) return
  // Move job to pending and place timeout
  const job = state.jobs.shift()
  state.pending.push({socket, job})
  // Setting job timeout
  setTimeout(() => {
    const index = state.pending.findIndex(el => el.socket.id === socket.id)
    if (index === -1) return
    // Remove pending job - TODO proper results handling
    state.pending.splice(index, 1)
  }, defaults.timeout)
  // Finally sending the job
  socket.emit('job', job)
}
// Simple interval for running 'dispatchJob' every X seconds
setInterval(() => dispatchJob(), despatchLoop || 500)

Even if dispatchJob seems complex, basically it just moving jobs from one array to another jobs->pending->results, the only thing missing from this code example a function for adding a job to be dispatch:

// Adding new job to the jobs array
const addJob = (id, type, data) => jobs.push({id, type, data})

Client

On the client side we the code is much simpler and deals with worker registration and job consumption.

<!doctype html>
<html>
  <head>
    <!-- served by socket.io -->
    <script src="/socket.io/socket.io.js"></script>
  </head>
  <body>
    <script>
    // Connecting to the server with socket.io
    const socket = io(window.document.location.origin)
    // TODO - real processing
    const processJobPromise = data => Promise.resolve({value: 'OK'})
    // 'connect' - fired upon socket connection 
    socket.on('connect', () => {
      // registering the client as a worker (with custom type)
      socket.emit('register', {type: 'my-worker-type'})
    })
    // The process function
    const runJob = ({id, type, data}) =>  
      processJobPromise(data)
        .then(result => socket.emit('result', {id, type, result}))
        // TODO - proper error handling and notification
        .catch(console.error)
    // listen for jobs send over the socket
    socket.on('job', job => runJob(job))
    </script>
  </body>
</html>

• connect - fires on socket connection, we're emitting the register event here to register the client/socket as a worker.

• job - this event is emitted by the server with the data for job processing, upon finishing the job we emit the result event with the result.