HPC Super Computing for Mortals

The US has recently wrestled back the HPC Super Computer Crown from China with Summit, a 200 PetaFLOP Goliath with over 9000 22-core PowerPC CPUs and a mind-boggling 27,000 NVIDA Tesla V100 GPUs.

These huge computing resources are usually built for specific numeric computationally expensive problems. In the case of Summit, its main use case is Nuclear Weapons Simulation, though interestingly, this isn't one of the use cases they advertise on their website!

All modern High Performance Computing (HPC) architectures are of the basic design shown below.

The challenge is how to build a software framework to distribute and parallel process code over such an hardware architecture.

Historically, the main framework used has been Message Passing Interface, or MPI. In MPI tasks are distributed over the HPC architecture and communicate with each other via messages. The challenge with MPI is that the Software Engineer has to make decisions on how to split the tasks, the message interface and what type of network model the tasks will be run. These network models are broken down into:

• Point to Point - where two tasks send messages to each other directly
• Broadcast - where data is published to all tasks
• Scatter - where data is broken into partitions and distributed to multiple tasks
• Gather - essentially the reverse of scattering, where a single task gathers data back from multiple tasks
• Reduce - where a single task processing the data  from each remote task. This is very much a coordinated version of Point to Point

For simple parallel problems, such as algorithms which can be classed as embarrassingly parallel, the design choice can be relatively simple. However, as the set of algorithms and processing pipelines becomes more complicated, the MPI implementation becomes challenging. Often, with MPI, more time can be spent on the 'plumbing', rather than writing the code the solve the domain problem. It's no surprise that HPC software development with MPI is a very specialist skill and the advantages of large scale computing is out of reach for the 'average' Software Engineer or Data Scientist.

This is where the Python Dask library comes to the rescue. Dask excels in that it provides a familiar Python Numpy and Pandas like interface for common numeric, scientific and data science problems, wth a 'bag' API for more general purpose map./reduce like computation, suited to unstructured data. The real power of Dask, though, comes from the fact that it builds the most optimum task graph for you, so you can concentrate your effort on solving the domain problem, not how to maximise the resources of the HPC.

Lets see some code. Here's how you get a Dask Cluster started

from dask.distributed import Client, progress
client = Client(processes=False, threads_per_worker=4,
                n_workers=1, memory_limit='2GB')
client

Once spun up and available, Dask should return a message telling you the number of Cores and Memory the cluster has allocated. If you're using a shared HPC Cluster with other workloads running, you may not get all the Cores and Memory you request.

Lets create a large numeric array of data, using the familiar Numpy like syntax.

from dask.distributed import Client, progress
import dask.array as da
x = da.random.random((10000, 10000), chunks=(1000, 1000))
x
dask.array

Lets now carry out a simple computation on this array.

y = x + x.T
z = y[::2, 5000:].mean(axis=1)
z
dask.array mean_agg-aggregate, shape=(5000,), dtype=float64, chunksize=(500,)

Notice here that z has not returned an answer, but a pointer to a new Dask array. By default, all calls to the Dask API are lazy, and it's only when you issue a .compute() call that the whole preceding task graph gets submitted to the Cluster for computation and evaluation - like this.

z.compute()
array([0.99524228, 1.01138959, 1.00196082, ..., 0.99702404, 1.00168843,
       0.99625349])

The power of Dask is that you can get started on a Laptop with Python, then transfer your algorithm to a HPC Cluster and scale up the computation with no change to the code.

Ah, you'll say, I don't have access to a HPC like Summit. True, most people don't have access to a HPC Cluster. I'm personally lucky that I work for a company that has it's own private HPC environment. Unless you work in Academia, or for a large industrial company, typically in the Automotive, Aerospace / Defence and Pharmaceutical industries, you unlikely to be able to access that level of compute.

This is where Cloud Computing comes to the rescue. In particular services such as Amazon EC2 Spot Instances.EC2 Spot Instances allow to you request compute resources at substantial On-Demand discounted rates. This is because Amazon have the right to interrupt and pause your compute with only 2 minutes notice. For example, at the time of writing this article, you can have a m4.16xlarge (64 vCPU 256GB RAM) Spot Instance at ~$1 per hour - which is incredible. However, this particular configuration comes with a potential interruption rate of greater than 20%. However, if you optimise your compute workload in Dask to suit, for example you work all 64 vCPUs, you may never see any interruptions ~80% of the time.

So there you have it. Super Computing is now available for everyone. All you have to do is work out what major massive computation problem to solve. I'd recommend looking at this video for inspiration.

Further info can be found:

• Dask - https://dask.org/, https://blog.dask.org/
• Amazon EC2 Spot Instances - https://aws.amazon.com/ec2/spot/

The Freedom from Locks

I'm currently working on a project where I need to (i) cope with very high data rates over a shared memory buffer and (ii) squeeze as much processing power out of the (pretty low power) CPU as I possibly can. Oh, it it's going to have to be multi-threaded.

The system is on a POSIX platform and I experimented with Queues, Shared Memory but non of them gave the performance I needed. One of the big issues is making the application thread safe, and that usually involves locks. Locks are expensive and are, on most compilers and CPU targets, pretty slow. There is an interesting article here comparing performance. Locks are costly to acquire and release, and there's always the potentially for deadlock and contention scenarios.

So, I looked to adopting a Lock Free Ring / Circular Buffer approach. The challenge is to create a data structure and algorithm that allows multiple threads pass data between them concurrently without invoking any locks.

The diagram below shows the structure of a Ring Buffer.

This structure is often used in the classic multi threaded Producer / Consumer problem, but there are potential concurrency problems with this structure:

• The Producer must find free space in the queue
• The Consumer must find the next item
• The Producer must be able to tell the queue is full
• The Consumer must be able to tell if the queue is empty

All these operations could incur a contention / lock issue. So how to you get around this?

Firstly you need to define a static fixed global data structure with free running counters:

static volatile sig_atomic_t tail=0;
static volatile sig_atomic_t head=0;
static char buffer[CAPACITY];
static int mask=CAPACITY-1;

The Volatile keyword in C/C++ tells the compiler that the value held by the variable can be modified by another thread, process, or even external device. In fact the Volatile keyboard is often used to detect data from an external piece of hardware that uses memory mapped i/o.

sig_atomic_t is an Integer data type that tells the compiler to ensure that the variable is not partially written or partially read in the presence of asynchronous interrupts. Essentially used for signal handling in multi threaded / process contexts.

The combination of a volatile sig_atomic_t integer is that it creates a inter process thread safe signal handling variable that ensures any operation completes in a single CPU cycle that cannot be interrupted.

So, we've declared tail and head as our read and write positions in our circular buffer. Now, how do we insert a piece of data into the ring buffer?

int Offer(char e) {
 int currentTail;
 int wrapPoint;

 currentTail=tail;
 wrapPoint=currentTail-CAPACITY;
 if (head<=wrapPoint) {
  return 0;
 }
 buffer[currentTail % mask]=e;
 tail=currentTail+1;
 return 1;
}

Now lets retrieve a byte from the buffer.

char Poll(void) {
 char e;
  int i;
 int currentHead;

 currentHead=head;
 if (currentHead>=tail) {
  return 0;
 }
 i=currentHead & mask;
 e=buffer[i];
 head=currentHead+1;
 return e;
}

Now, here's Producer / Consumer thread that invokes our lock free buffer.

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/time.h>
#include "LockFreeBuffer.h"

#define REPETITIONS 10000
#define TEST_VALUE 65
#define ITERATIONS 256
#define BILLION 1E9

void *Producer(void *arg);
void *Consumer(void *arg);

int main(void)
{
 int i,t;
 double secs;
 struct timespec start,stop;
 pthread_t mythread[2];

 printf("Starting...\n");
 for (i=0;i<ITERATIONS;i++) {
  Init();
  clock_gettime(CLOCK_PROCESS_CPUTIME_ID,&start);
  if (pthread_create(&mythread[0],NULL,Producer,NULL)) {
   perror("Thread failed to create...\n");
   exit(-1);
  }
  if (pthread_create(&mythread[1],NULL,Consumer,NULL)) {
   perror("Thread failed to create...\n");
   exit(-1);
  }
  for(t=0;t<2;t++) {
   if (pthread_join(mythread[t],NULL)) {
    perror("Thread failed to join...\n");
    exit(-1);
   }
  }
  clock_gettime(CLOCK_PROCESS_CPUTIME_ID,&stop);
  secs=(stop.tv_sec-start.tv_sec)+((stop.tv_nsec-start.tv_nsec)/BILLION);
  printf("Operations/sec: %4.0f\n",REPETITIONS/secs);
 }
 printf("Completed...\n");
 exit(0);
}

void *Producer(void *arg)
{
 int i;

 i=REPETITIONS;
 do {
  while (!Offer(TEST_VALUE));
 } while (0!=--i);
 pthread_exit(NULL);
}

void *Consumer(void *arg)
{
 char buf;
 int i;

 i=REPETITIONS;
 buf=0;
 do {
  while (buf=Poll());
 } while (0!=--i);
 pthread_exit(NULL);
}

If you're interested in more on lock free buffers, check out the LMAX Distruptor for Java

Domain Specific Languages

If, like me, you started your software engineering career on classic time-sharing systems (for me it was a VAX 11/780) you probably had a small restricted set of programming languages available to you, e.g. Fortran, BASIC, Pascal etc. These languages are all classed as General Purpose Languages. What binds GPLs together is that they are designed to solve any class of software problem and are not tied to any domain. In fact, when you examine the structure of languages like C, BASIC, Fortran, Java, they all share a lot of common constructs in terms of control structures, branches, looping, variable assignment etc.

This was all straightforward enough, in fact early in my career I moved  (fairly) freely from programming in BASIC, Fortran, C and Pascal. In the early 90s I moved into the CAD (Computer Aided Design) / PLM (Product Lifecycle Management) industry, specifically the AutoCAD industry. It was at this point I was introduced to a language which, at first, completely foxed me. This language was Lisp, specifically AutoLisp.

At a first view, Lisp looked quite alien, endless parenthesis, strange functions such as car, cons and cdr, but once I got to grip with he basics the power of the language became apparent. The key to Lisp are sets and lists, in fact Lisp stands for List Processing. Here's a bit of sample Lisp code:

(setq cntr 1)
(while (cntr < 5)
 (princ cntr)
 (seqt cntr (+ cntr 1))
)

A pretty simple example that sets the variable cntr to 1, prints the value of cntr to the console, loops round 4 times. The key to the Lisp language it sets and lists and Lisp is fairly unique in that it is a homoiconic, i.e. the representation of the language syntax is a data structure primitive type of the language itself. XSLT and XQuery are also classed as homoiconic languages.

Lisp lends itself to programming AutoCAD quite well, as it was set and list orientated and with a CAD system a lot of the programming was aimed at manipulating geometry of the CAD model / drawing.

Anyway, back to homoiconic. The clever bit about Lisp is the fact that functions can be treat in the same way as variables, that is, functions can be passed as function arguments and returned as function values. It's this key property that allows powerful abstractions to be built in a language like Lisp, hence it use as a DSL. As always an example helps:

(defun double (x)
 (* 2 x) /* define a function double that simply times the argument by 2 */
)

Now, here's the clever bit, if we want to compute 2⁴ we could use the double transform on 1 four times.

(double (double (double (double 1))))

Now, that's a bit long winded, even better would be a function that took a function and an object and repeated the function transform n times. In Lisp, that's doable.

(defun repeat-transformation (F N X)
  /* Repeat applying function F on object X for N times */
  (if (zerop N) /* Returns true if the argument is zero */
      X
    (repeat-transformation F (1- N) (funcall F X))
   )
)

What's happening in this code is, of course, recursion. The key is the funcall which given a function F and objects X₁,X₂...X_n the form (funcall F X₁ X₂ ... X_n) invoke the function F with the arguments X₁ X₂ ... X_n. The variable N is a counter keeping track of the remaining number of times we need to apply function F to the accumulator variable X.

I'm not planning to do in-depth programming in Lisp here. If you're interested in learning more about Lisp I recommend you take a look at Practical Common Lisp by Peter Seibel. The text for the book can be found on-line here. Hopefully, though, you've got the idea that a programming language can, in essence, manipulate itself to create a new subset language, hence a Domain Specific Language.

So what use are DSLs? A good real life practical example is Apache Camel. Camel is an integration framework that provides a implementation structure and runtime for the classic Hohpe and Woolf Enterprise Integration Patterns (EIP). Integration is a great application for a DSL as you have a common patterns to solve and components to 'wire' together. In Camel's case, the common elements are wiring Processors that carry out functions such as transformation, mediation, routing etc, with Endpoints that allow messages to be sent and received, e.g. JMS, HTTP etc.

To demonstrate how productive and powerful a DSL (and in this case Camel) can be, imagine you've got a systems integration problem to solve which involves reading files from one directory and placing them in another. Simple right? Here's a possible solution in Java.

public class FileCopier {
 public static void main(String args[]) throws Exception {
  File inboxDirectory = new File("data/inbox");
  File outboxDirectory = new File("data/outbox");

  outboxDirectory.mkdir();
  
  File[] files = inboxDirectory.listFiles();

  for (File source : files) {
   File dest = new File(
    outboxDirectory.getPath()
    + File.separator
    + source.getName());
    copyFile(source, dest); 
  }
 }

 private static void copyFile(File source, File dest)
  throws IOException {
   OutputStream out = new FileOutputStream(dest);
   byte[] buffer = new byte[(int) source.length()];
   FileInputStream in = new FileInputStream(source);
   in.read(buffer);
   try {
    out.write(buffer);
   } finally {
    out.close();
    in.close();
   }
 }
}

There's a fair bit of code their for a simple task. Also, we haven't addressed error handling, concurrency, polling for the files, keeping track of what files have been moved etc etc.

Now here's the sample problem solved with Camel DSL.

public class FileCopierWithCamel {
 public static void main(String args[]) throws Exception {
  CamelContext context = new DefaultCamelContext();
  context.addRoutes(new RouteBuilder() {
   public void configure() {
    from("file:data/inbox?noop=true")
    .to("file:data/outbox");
   }
  });
  context.start();
  Thread.sleep(10000);
  context.stop();
 }
}

A lot simpler. Most of the code above is Camel boilerplate, for example setting a CamelContext that's started (context.start()) and subsequently stopped (context.stop()). The integration logic is all defined in the RouterBuilder function and the from.to function. Essentially down from around 30 lines of Java code to 14. Also, all the things we didn't address in the pure Java example, e.g. error handling etc, is all handled by the Camel framework and runtime.

For another good example of application of DSL is FIT (Framework for Integrated Test). FIT looks to solve that eternal software engineering problem of testing and test scripts and closing the gap between developers, business analysts and end users. FIT reads HTML tables that map onto system classes methods and properties. A Business Analysts or End User creates a FIT document that describes the tests to be carried out in the form of the table with inputs and expected outputs. The FIT document can be created in Microsoft Word and exported as HTML. Developers then create Fixtures which map the FIT documents onto the business logic code of the application being tested.

In the case FIT, the DSL is, what FIT term, the Fixture. Fixtures wrap the specific application business logic that allows the FIT documents to execute the tests. A simple Fixture class can be seen below.

import fit.ColumnFixture;

public class Division extends ColumnFixture { /* Extending the ColumnFixture class allows the FIT runtime to map the input test tables to the logic code */
    public float numerator;
    public float denominator;
    public float quotient() {
        return numerator / denominator;
    }
}

The language that's currently getting most attention in the DSL world is Groovy. Groovy, if you've not yet come across it, is a lightweight Java like language and, like Lisp, supports closures and code as data and, therefore, ideally lends itself to DSL creation. There is a great simple example of a DSL for Stock Market transactions written in Groovy by Justin Spradlin here.

Increasingly, there are tools becoming available for DSL construction. A good example of one of these is Meta Programming System (MPS) from JetBrains.

So what's the future for DSL? I believe they have a place. I'm not sure there are too many advantages in trying to build DSLs for business applications, they are just too varied to gain extensive reuse and may be too unfamiliar for use by end users. The Apache Camel project shows the way for me with DSLs, accelerating the development of common development patterns, Groovy is also being used for this kind of 'plumbing' application.

If you want to found out more about DSLs I'd recommend visiting Martin Fowler's Wiki and taking a look at his book Domain Specific Languages.

Programming with Eclipse BIRT

A couple of years ago I started working on an Agent-Based Simulation System. This system simulated the behaviour of fleets of complex assets, such as aircraft, to support predictive analysis of maintenance, availability, system failure etc.

One of the (many) key challenges of the system architecture was the shear volume of data each simulation could potentially generate and the visualisation of the results. At the time the organisation I was working in hadn't decided on a corporate Business Intelligence / Analytics solution, so I had the task of looking for an appropriate technology to plug in to the system architecture.

It made sense that the solution was Open Source, looking around there were a few options including Jasper Reports, in the end I decided on Eclipse BIRT. I choose BIRT mainly because we were already using the Eclipse Platform for the rest of the system architecture development effort and, to be honest, I was quietly impressed with the capability, power and ease of use. Also BIRT has the advantage of being Open Source but with a solid commerical entity behind it in the form of Actuate.

BIRT's architecture is, essentially, split into two components, a Designer tool to maintain the report design, and a Java runtime Report Engine that manages and executes the report. The Designer component can either run as an Eclipse RCP client or Eclipse IDE Plug-In. The Report Engine will deploy to any standard Servlet Engine / J2ee Application Server environment, e.g. JBoss AS, Tomcat, Websphere etc.

An overview of the architecture is shown below.



The architecture of BIRT is pretty open and extensible. For example data source connectivity is provided by ODA which allows you to connect to different physical data sources, such as JDBC RDBMS, Web Services, Flat Files or Java Beans, yet treat them all as logic result sets. In fact, one of my first 'evaluations' of BIRT's capabilities involved building a web reporting front-end to Amazon's ECS (now known as AWS Associates Web Service).

For me, the real power of BIRT is the Server Event model that allows you alter the look, feel or behaviour of the report at runtime. An API is provided for the event model and can be accessed via Rhino (Javascript based scripting for Java) and native Java. An overview of the event model can be seen below.


Events are exposed on numerous report objects, for example DataSet.beforeOpen(), ReportElement.onPrepare() etc.

These events allow you to do all sort of 'clever stuff' at runtime. For example, change the SQL of a query based on a parameter or the outcome of a previous query, change repot or chart colours based on calculated values.

For example, the following code changes the SQL expression dependent on a runtime parameter rp_sortorder:

myDataSet.beforeOpen() {
  this.queryText=this.queryText.replace('sortlistorder',params["rp_sortorder"]);
}

You can also get access the the J2EE runtime container. For example, the following code gets access to a Http Session object.

report.initialize() {
  importPackage(Packages.javax.servlet.http);
  var thisSession=reportContext.getHttpServletRequest().getSession();
  var mySessionObject=thisSession.getAttribute("mySessionObject");
  var mySessionObjectValue=mySessionObject.getValue();
}

As well as a Report Design and Runtime (Engine) API, there's a Chart API that allows to extend the built in Chart functionality. For example, the code below alters the scale of a chart dependent on two global variables "ymindate" and "ymaxdate".

function beforeGeneration(chart, icsc) {
  importPackage(Packages.org.eclipse.birt.chart.model.data.impl);
  var scriptObj=icsc.getExternalContext().getScriptable();
  var yaxisminDate=scriptObj.getPersistentGlobalVariable("yminDate");
  var yaxismaxDate=scriptObj.getPersistentGlobalVariable("ymaxDate");
  var yaxismin=Date.parse(yaxisminDate);
  var yaxismax=Date.parse(yaxismaxDate);
  var xaxis=chart.getBaseAxes()[0];
  var yaxis=chart.getOrthogonalAxes(xaxis,true)[0];
  var yaxisScale=yaxis.getScale();
  yaxisScale.setMin(DateTimeDataElementImpl.create(yaxismin));
  yaxisScale.setMax(DateTimeDataElementImpl.create(yaxismax));
}

I am impressed with BIRT and I haven't yet found a reporting / analytics requirement it can't handle, coupled with it's extensibility and the backing of the Eclipse community and Actuate I highly recommend it.

If you want to find out more about BIRT:

• BIRT Eclipse - the home of BIRT development
• BIRT Exchange - a well supported and active BIRT community site
• BIRT BlogSpot - a blog about BIRT run by key BIRT team members