DEV Community: Peter Ohler

The Pretty JSON Revolution

Peter Ohler — Tue, 23 Feb 2021 00:11:46 +0000

Having a bit of fun with JSON formatting. http://www.ohler.com/dev/pretty.html

Building Solid Go GraphQL Applications Quickly

Peter Ohler — Tue, 02 Feb 2021 19:40:34 +0000

Black box testing is a great way to build GraphQL applications quickly. Check out a step by step with code that can be reused here.

Testing Asynchronous Threads in Ruby

Peter Ohler — Tue, 06 Nov 2018 13:56:33 +0000

Threads and asynchronous environments are initially a bit tricky. Without a good mental model to organize interaction, it is easy to get into trouble and end up with unexpected results. On top of that, testing asynchronous code can be difficult without the right tools or test patterns.

Thinking about threads as people and shared objects as ‘things’ that can be owned helps to organize the working of a multithreaded system. In this episode we will go through an example to learn all about testing asynchronous Ruby code.

If you are using Rails or Rack or really any application that as a web browser front end you are in an asynchronous environment. The Rack #call is always called asynchronously. So whether you know it or not, there is a good chance you are already using multithreaded components.

Testing: Trigger, collect and check

Testing an asynchronous callback API can be made synchronous by following a pattern of three steps; trigger, collect, and check. Think if each thread as a separate individual and objects as, well, things that can be owned by just one individual at a time.

We’ll use the example of Batman and his 7 different suits. Because that is a practical example and we can understand the importance of knowing whether all suits are with Alfred to be washed when you are about to run out and save the city.

Example: Laundry day at the batcave

The example is Alfred washing Batman's suits. The SuitWashScheduler is a scheduler that invokes a callback for each washing event. The scheduler makes seven callbacks at one-second intervals starting one second after starting. The trigger is the
creation of the SuitWashScheduler.

class SuitWashScheduler
  def initialize(cnt)
    Thread.new {
      cnt.times {
        sleep(1.0)
        yield
      }
    }
  end
end

Collecting

Collecting results must be thread safe to avoid race conditions. Any object shared across more than one thread has to be protected. Protection is a way of keeping track of the owner of an object. Only the owner can make changes or view the object. A suit can only be with Batman to use in fights, or with Alfred to be washed.

To stay friendly a thread (in the metaphor Batman or Alfred) only takes ownership for short times and then gives up the ownership. A Mutex is usually used to keep track of the owner. The SuitwashScheduler callback will own the result counter when the counter is incremented. The callback that is run in the SuitWashScheduler thread signals that all results have been received when the counter hits the target.

Writing the example starts with setting up some globals. In a real application the globals would be replaced by class or object attributes.

$main_thread = Thread.current
$mu = Mutex.new
$count = 0
$target = 7

Management and owners

The $main_thread and $mu are used to manage the threads and waiting for completion of the test while the $target and $count track the test results. Remember this is a trivial test so collecting and checking results has to be simple.

The test is started by creating a new instance of the SuitWashScheduler, giving the initializer the $target number of iterations. In this case, the 7 suits that need washing.
The block provided will be run in the SuitWashScheduler thread. For each iteration the $count is incremented and printed.

Looking ahead we realize that the main, test thread is going to be checking the$count also which means it will need ownership of the $count as well so a means of taking ownership of $count is needed. The $mu Mutex instance is the ownership token. In the block passed to the SuitWashScheduler.new call a $mu.synchronize block takes ownership long enough to set the $count and check the results. More on the results check in a moment.

SuitWashScheduler.new($target) {
  $mu.synchronize {
    $count += 1
    puts $count
    $main_thread.wakeup if $target <= $count
  }
}

Check: are all suits done?

Back at the main thread we need to wait for the test to finish. Batman needs to wait before all 7 suits are done. There are two conditions to check for; the tests updates the $count as expected or Batman gets bored waiting for the test to finish and times out. Before checking the $count to see if it has reached the $target, the ownership of $count is needed. Just like in the block for the SuitWashScheduler a call to $mu.synchronize is used.

But that can't be right, if we lock up the main thread how can the SuitWashScheduler thread ever change the $count? Luckily for us there is a neat trick that takes care of this. The Mutex class has a #sleep method that gives up ownership and waits until either it times out or is woken. Once woken either through the timeout or a #wakeup call to the main thread $mu attempts to take ownership again before continuing. Once ownership has been achieved the results can be checked and a determination of the pass or fail state of the test can be made.

$mu.synchronize {
  $mu.sleep($target + 1)
  if $target != $count
    puts 'FAILED'
  else
    puts 'Passed! All suits are washed and clean'
  end
}

If you want to get deeper into this, you can make the example a bit more interesting by trying to create multiple schedulers and see how the Mutex keep the $count changes from colliding. As if Batman sends some suits to Alfred to be washed, and some others to the dry cleaners. Make sure to change the logic to make sure the $target check is a total of all the expected yields.

Roundup

Working with threads and asynchronous environment gets easier with the right mental model.
In this post we used people as a metaphor for threads and physical objects (suits) as metaphor for shared objects, that can only be owned by one thread or person at a time. We think this way to abstract makes it easier to understand and remember.

We hope the examples with make you remember the mechanisms of async, but we hope the image of Batman running out while all his suits are being washed won’t stick with you for too long.

PS If you are done with all the batman metaphors on the blog, let us know.

Peter Ohler creates quite bit of a high performance code, and writes about it too, every now and then. He made the Agoo gem, which is a pretty cool high performance HTTP server.

Push and Pub/Sub in Ruby

Peter Ohler — Wed, 29 Aug 2018 05:37:39 +0000

Building real time features in Rails has gotten a lot easier to do with libraries like Action Cable. In this episode of AppSignal Academy, we'll dive into real time updating and toy around with building a minimal WebSocket server to look at how it works under the hood.

We'll be building an application that pushes data, and uses Pub/Sub over a WebSocket. Before we start on the code, let's first spend a little time covering what those three concepts mean:

Push refers to pushing data to a receiver instead of having the receiver poll for that data. A must have in realtime updates like stock prices, chat applications, or operations consoles.
Pub/Sub or publish and subscribe is an interaction model for pushing data made popular by TIBCO on Wall Street in the 1990s. A receiver subscribes to a subject and waits for a publisher to push data to that subject. It is common to include wildcard pattern matching to match a published message to a listener, although some simpler implementations only use named channels instead of wildcards in the subject. I got started at TIBCO in those early days so I like the flexibility of the wildcard pattern matching.
WebSocket is a protocol for exchanging data - usually between a web browser and an application. An HTTP connection is upgraded to a WebSocket connection and then data can be sent both ways between the two endpoints. WebSockets can push data from an application to the browser. It also provides another mechanism other than a POST or PUT for sending data from your JavaScript code in the browser back to the application. Pretty nice, huh?

Under the Hood

Let's look at how an example of a WebSocket server could work. From a browser, the client tries to make a WebSocket connection to a server with JavaScript code.

var sock = new WebSocket("ws://" + document.URL.split('/')[2] + '/upgrade');

The server receives an HTTP request with an indicator that an upgrade was requested. Generally, the server lets an application decide on whether to upgrade or not. How it does that depends on the API provided to the app. A server that supports Rack provides an option to hijack the socket and let the developer handle all the protocol details or, according to a proposed PR, a response to upgrade is enough.

The upgrade is a set of exchanges between the server and the client. All browsers and some server gems hide these details. Once the connection is established, messages can be exchanged following the WebSocket protocol.

The magic under the hood handles encoding, decoding, and the message exchange protocol. Messages are binary, fixed-width structures with a trailing payload, encrypted using SHA1. The WebSocket protocol includes several message types and exchanges, like ping/pong heartbeats and opening and closing message exchanges. That's the magic the servers perform by not using the connection-hijack approach.

Diving In

We'll use the example of a clock thread that is started to publish the current time to all listening clients. We'll use Agoo to build our server because it's fast and keeps the complexity to a minimum.

We will start out with some JavaScript as a client by showing the current time on an HTML page. After creating a new WebSocket an onopen callback is set that changes the status HTML element. The onmessage callback updates the message HTML element. Callbacks are a common design pattern when working with asynchronous calls such as publish and subscribe exchanges.

<!-- websocket.html -->
<html>
  <body>
    <p id="status"> ... </p>
    <p id="message"> ... waiting ... </p>

    <script type="text/javascript">
      var sock = new WebSocket("ws://" + document.URL.split('/')[2] + '/upgrade');
      sock.onopen = function() {
          document.getElementById("status").textContent = "connected";
      }
      sock.onmessage = function(msg) {
          document.getElementById("message").textContent = msg.data;
      }
    </script>
  </body>
</html>

With our client done, let's implement the server, which is a Ruby application using the Rack API. The Clock class itself will be a handler for all HTTP requests on the /upgrade path. If the request is for an upgrade we simply return Success with an HTTP status code of 200, otherwise, we return a 404 for Page Not Found. The only other step in the #call method is the assignment of the WebSocket handler.

class Clock
  def self.call(env)
    unless env['rack.upgrade?'].nil?
      env['rack.upgrade'] = Clock
      [ 200, { }, [ ] ]
    else
      [ 404, { }, [ ] ]
    end
  end
end

The API is based on callbacks. The only callback we care about for our server is the #on_open callback which enables us to create a subscription to the "time" subject. Messages are exchanged over channels identified by subjects or topics. #on_open is called when a web socket connection is established.

class Clock
  # ...

  def self.on_open(client)
    client.subscribe('time')
  end
end

Now, let's start publishing with a thread that publishes the time after every second. The call to Agoo.publish sends a message on the "time" subject, then all subscribers receive the message. The server keeps track of the subscriptions and connections and delivers the message to the JavaScript client which updates the HTML element.

require 'agoo'

Thread.new {
  loop do
    now = Time.now
    Agoo.publish('time', "%02d:%02d:%02d" % [now.hour, now.min, now.sec])
    sleep(1)
  end
}

The only other code needed is the code that initializes and starts the server. The call to Agoo::Server.handle(:GET, '/upgrade', Clock) tells the server to listen for HTTP GET requests on the /upgrade URL path and to pass those requests to the Clock class. This allows the routing to occur outside Ruby for improved performance and flexibility.

Agoo::Server.init(6464, '.', thread_count: 0)
Agoo::Server.handle(:GET, '/upgrade', Clock)
Agoo::Server.start

We are almost there. Run the server with this command.

$ ruby pubsub.rb

A log entry should appear showing something like the following, indicating that the server is running and listening on port 6464.

I 2018/08/14 19:49:45.170618000 INFO: Agoo 2.5.0 with pid 40366 is listening on http://:6464.

Time To See If It's Working

Let's open http://localhost:6464/websocket.html. After an initial flicker as a connection is established, the connection status and the time should be displayed. The time will increment every second as the clock ticks.

connected

19:50:12

Congratulations on making a publish and subscribe web application ;-)

In today's episode, we looked at using WebSocket. Server Side Events (SSE) offers another option of doing the same, and we have included SSE in the full source code example. If you want to find out more, take a look at the Agoo server we used or Iodine WebSocket Server.

We'd love to know what you thought of this article, or if you have any questions. We're always on the lookout for topics to investigate and explain, so if there's anything magical in Ruby you'd like to read about, don't hesitate to leave a comment.

This post is written by guest author Peter Ohler. Peter creates quite bit of a high performance code, and writes about it too, every now and then. He made the Agoo gem, which is a pretty cool high performance HTTP server.