How to deal with blocking operations

In distributed environment even simple things become complicated. One of the most trivial examples is a counter. If you have only one instance of an application running, then you can simply store the counter in a database table with the counter's id and value, for example:

| ID | VAL |
------------
| 0  |  1  |

and increase its value with a simple query:

UPDATE counters SET val=val+1 WHERE id=0

However, when there are many clients trying to increase the counter simultaneously (for example multiple frontends displaying the number of visitors to the website), you may experience some serious slowdowns. It's because updating a row in a database is a blocking operation, which means that during update the row is locked to prevent the situation when multiple process try to modify it at the same time. There are two basic types of locks: write locks and read locks. A write lock is a basic and most often used type of lock, which does not allow modifying a value while another change is in progress. You can still read the row value, but in this case two subsequent reads can return two different values, because the change could have taken place just between them. When this becomes an issue, you can use a read lock, which prevents other processes from reading the row until the change is done. Depending on how your database is configured, locking the counter row will become more or less painful, but it still needs to be done sequentially to ensure its value is correct. This way, simple operation of increasing the visitors counter can become a serious bottleneck to the whole website.

I have prepared an example application which simulates such case. You can download a zipped archive from here. It was prepared in Linux, but in case you work in Windows you should also be able to compile it with Cygwin - just make sure to install mysql development libraries (you need mysql_config.exe binary in your PATH).
First, prepare the database and set up the tables. All example assume that you have a MySQL database running on localhost, with test database called "test" and password "root" for the root user. If your environment differs, modifiy respective variables:

mysql -u root -proot test < counter.sql

Now, compile the example code which works on the counter:

make sync

This example simulates ten simultaneous blocking counter updates. Each transaction starts one second after the previous one, but it takes two seconds to complete, so the transactions overlap and block one another.
When you run the resulting binary you should see the output similar to the following:

Begin #1
Update #1
Begin #2
Update #2
Begin #3
Update #3
Commit #1
Begin #4
Update #4
Begin #5
Update #5
Commit #2
Begin #6
Update #6
Begin #7
Update #7
Commit #3
Begin #8
Update #8
Begin #9
Update #9
Commit #4
Begin #10
Update #10
Commit #5
Commit #6
Commit #7
Commit #8
Commit #9
Commit #10
Counter value: 10, time elapsed: 20.015499s

You can notice that commits do not take place immediately after updates. This is because each transaction must wait for all previous ones to complete. As a result the test takes around 20 seconds, which is the sum of time needed to finish all transactions. In this scenario, the more clients try to modifiy the counter, the longer it takes to complete. Finally, the system becomes overloaded, some of the operations time out, client processes fail to set the correct counter value, and the user experience is a disaster.

There are couple of ways to solve the problem. Some of them involve spending time and money to build a NoSQL cluster based on some fashionable technology, like Hadoop or Cassandra. A cheaper and smarter way involves implementing so called "optimistic locking". The main idea behind this technique is to decrease to the minimum the time you spend in lock. Operation based on optimistic locking takes three phases:

• read the value
• modify the value
• compare and swap

The last operation is the only one that needs to be atomic. In short, it reads the value you want to modify and if didn't change from the first read, it replaces the old value with the new one. If comparison fails, it means that another process has already modified the value, and we need to roll back.
Optimistic locking is a good solution, but it still can lead to failures while updating counter value, and it also requires clients to implement some strategy on how to deal with update failure: "Should I repeat the operation?", "At what intervals?", "How many times before giving up?", etc.

Another way is to get rid of blocking opeartions at all, and replace them with non-blocking ones. It can be done easily by introducing queues. Putting a new value into a queue does not affect any existing values, so it doesn't need any read or write locks. With every new request you just insert a new row into the queue table, and there is a single, separate process which updates the counter and removes the rows it has read form the queue. Because there are no concurrent processes working on the counter, there is no problem of blocking. Also, the worker can read new inserts in batches instead of single records, which can give a huge performance boost with fast growing queue (if there are a hundred new tasks waiting in the queue, you can increase the counter by a hundred in one step). The queue is safe, because each insert has its own unique id, so deleting records from the queue does not interfere with inserting new ones. Also, when updating fails, the inserts are not lost, but they remain in the queue for later processing. The only drawback is that when you read the counter value it is still a little bit behind, but for purposes such as the number of visitors to the website, it can be easily accepted.

Going back to the example. Compile the code with:

make async

This will rebuild the code to use queue instead on changin the counter directly. When you now run it, you should see the output similar to the following:

Begin #1
Insert #1
Begin #2
Insert #2
Commit #1
Begin #3
Insert #3
Commit #2
Begin #4
Insert #4
Begin #5
Insert #5
Commit #3
Begin #6
Insert #6
Commit #4
Commit #5
Begin #7
Insert #7
Commit #6
Begin #8
Insert #8
Commit #7
Begin #9
Insert #9
Commit #8
Begin #10
Insert #10
Commit #9
Commit #10
Counter value: 20, time elapsed: 11.023697s

As you can see, the whole operation now took much faster to complete, because the new transactions didn't have to wait for the previous ones to finish.

Scale up your skills, not the hardware

It is not a big secret that good programmers keep their saws sharp all the time. If you want to improve your mental skills a good place to start is Project Euler. I found it while browsing other programmers' blogs and immediately got hooked up. Project Euler is a place where you can challenge yourself (and others) in solving various mathematical and computer programming problems. The collection of problems is constantly growing and you can find them ranging from very easy to really difficult. When you provide the correct solution to a problem, a forum dedicated to it becomes revealed and you can see how other people managed to solve it. Sometimes you can also get a bonus in a form of paper explaining the problem and the best known solutions in details. You can learn a lot from these forums since people share there both their ideas and source code written in different programming languages; some of them don't even use a computer, all they need to find a solution is paper and pencil.

What I particularly like about Project Euler is that most of the problems cannot be solved by brute force. They can make you check hundreds or thousands of possible combinations, but still require a more clever approach. A classic example are problems 18 and 67, where you need to find a path from the top of the number pyramid to the bottom, so that the total sum of numbers on the path is the largest. While the pyramid in problem 18 is rather small and you can find the solution by trying every of 16384 possible paths, it is impossible to solve problem 67 the same way. To solve it you need to turn the problem upside down - literally, since you can easily get the correct solution if you start searching for the best path from the bottom of the pyramid to the top, reducing the number of combinations with every step.

You may wonder how mathematics can help you in facing every day challenges as a programmer. Imagine you work for an on-line store with millions of CD records on stock, and you provide a browser with a dynamic list of products based on different criteria, such as genre, year and price. Now you want to show to your customers the most popular product within the range they choose. Getting all items satisfying the selected criteria from a database and finding the most popular one every time someone browses the list will quickly kill your back-end. On the other hand, it is impossible to generate such data in advance for every possible combination of search parameters. You can either expand your server farm (and possibly move to the cloud to reduce costs) or you can use some statistical tools to help you do the job. The trick is to find a representative sample size for a given population of products. Assuming that N is a total population, z is a statistical factor for desired confidence level, p is an estimated proportion of a given variable within the population, and d is admissible error level, we can get sample size n with the following formula:

x = z^2 * (p * (1 - p))
n = N * x / ((N - 1) * d^2 + x)

The z factor is constant and equals 1.96 for the 95% confidence level and 2.58 for the 99% confidence level. Assuming that a selected range of CD records contains one million uniformly distributed products and you can tolerate a 5% error, it turns out that in 95% cases you get the most popular one correctly by analysing only 385 randomly chosen items from the list, since:

x = 1.96^2 * (0.5 * (1 - 0.5)) = 0.9604
n = 1000000 * 0.9604 / ((1000000 - 1) * 0.05^2 + 0.9604) =~ 384.013

To be 99% sure that no more than 1% results are incorrect, you'd need to analyse 16369 records.

You can learn more about calculating sample sizes from this article at Quirks - I took the above formula from it. And remember: while it's true that machine cycles are cheap and people cycles are not, never forget that people cycles are worth billions of machine ones.

MapReduce with Gambit Scheme & Termite

Termite is a library that runs on top of Gambit Scheme and implements concurrency model inspired by Erlang (there is also a Chicken Scheme port, but it's so far incomplete). Since I have already posted about Termite some time ago, I will not go into details how to write distributed applications with it. Instead, I want to show you how to use Termite to implement a simple MapReduce algorithm.

A trivial implementation uses a simple worker process that executes some code, and a pmap function which spawns multiple workers and collects the results:

(define (worker)
  (let* ((msg (?)) (pid (car msg)) (fun (cadr msg)) (arg (caddr msg)))
    (! pid (fun arg))))

(define (pmap fun lst)
  ; spawn workers
  (for-each 
    (lambda (x) (let ((p (spawn worker))) (! p (list (self) fun x))))
    lst)
  ; collect results
  (let loop ((i (length lst)) (result '()))
    (if (= 0 i) (reverse result) (loop (- i 1) (cons (?) result)))))

So far it works similar to its Erlang equivalent. However, this version does not guarantee (just like the Erlang one) that you get the results in the same order as the arguments passed to pmap. Let's consider function f, which holds execution of the current thread for a given number of seconds:

(define (f x) (thread-sleep! x) x)

If you run the following code:

(pmap f (list 1 2 3 2 1))

The result will be:

(1 1 2 2 3)

Since the first results on the list will come from the processes that finished first.

A solution to this problem is to make a single worker process using a sequencer to mark spawned threads. Numbers generated by the sequencer are then used to sort the results to achieve the correct order of the result list:

(define (worker)
  ; init sequencer
  (let loop ((n 0))
    (let ((msg (?)))
      ; terminate worker on request
      (if (not (eqv? 'exit (car msg)))
        (let ((pid (car msg)) (fun (cadr msg)) (arg (caddr msg)))
          ; start a new thread
          (spawn (lambda () (! pid (cons n (fun arg)))))
          ; increase sequencer
          (loop (+ n 1)))))))

(define (pmap fun args)
  ; start worker
  (let ((p (spawn worker)) (result '()))
    ; send data to the worker
    (for-each (lambda (x) (! p (list (self) fun x))) args)
    ; collect results
    (let loop ((i (length args)) (lst '()))
      (if (= 0 i)
        (set! result lst)
        (loop (- i 1) (cons (?) lst))))
    ; terminate worker
    (! p (list 'exit))
    ; sort results
    (let loop ((in (qsort result <)) (out '()))
      (if (null? in)
        out
        (loop (cdr in) (cons (cdar in) out))))))

You can use any function you wish to sort the results. I used a modified version of quicksort found at Rosetta Code:

(define (qsort l gt?)
  (let q ((l l))
    (if (null? l)
      l
      (let ((s (split-by (cdr l) (lambda (x) (gt? (car x) (caar l))))))
        (append (q (car s)) (list (car l)) (q (cdr s)))))))

(define (split-by l p)
  (let loop ((low (list)) (high (list)) (l l))
    (if (null? l)
     (cons low high)
     (if (p (car l))
       (loop low (cons (car l) high) (cdr l))
       (loop (cons (car l) low) high (cdr l))))))

Now when you run the test again, you get the result list in the same order as the argument list, no matter what the execution time of single processes was:

(pmap f (list 1 2 3 2 1))

(1 2 3 2 1)

You can download the pmap library here.

Using trampolines in C

In my previous post I presented my implementation of trampoline code in newLISP. Since this technique is also used in non-functional programming languages, I prepared an example in C.

Let's assume we have two mutually recursive functions f1 and f2:

#include <stdio.h>

void f1(int n);
void f2(int n);

void f1(int n) {
  printf("%d\n", n);
  if (n == 0)
    printf("Blastoff!\n");
  else
    f2(n);
}

void f2(int n) {
  f1(n-1);
}

int main() {
  f1(1000000);
}

When you compile this code without any optimizations provided by a compiler and try to run it you will most probably quickly get a segmentation fault (or similar error, depending on your platform). What happens here is the application running out of stack for storing function callbacks.

In Lisp you can solve this problem by rewriting functions to return continuations instead of values. In C you can simulate continuations by the following means:
1. Making functions to operate on pointers to variables instead of actual variables, which allows the application to preserve their values between function calls.
2. Returning a pointer to the next function to be called instead of calling it directly, which makes the use of application stack unnecessary.

So rewritten f1 and f2 can be presented in the following form:

#include <stdio.h>

typedef void *(*Fun)(int *n);

void *f1(int *n);
void *f2(int *n);

void *f1(int *n) {
  printf("%d\n", *n);
  if (*n == 0) {
    printf("Blastoff!\n");
    return NULL;
  } else {
    return f2;
  }
}

void *f2(int *n) {
  *n = *n - 1;
  return f1;
}

int main() {
  int n = 1000000;
  Fun f = f1;
  while (f != NULL) {
    f = (Fun)f(&n);
  }
  return n;
}

Now the main function iteratively goes through subsequent function calls until the halting condition (NULL) is met. At the end of iteration variable n holds 0.

As a side note, gcc is able to optimize the code of mutually recursive functions quite effectively. If you compile the first example with gcc -O2 the resulting code will not crash, because the compiler replaces all call instructions with jmp.

P.S. I would like to thank Johnny, my friend and a true programming expert, who inspired me to write this post. I was also inspired by Carlos Oliveira's post on trampolines.

Advanced recursion in newLISP

In my recent post about newLISP I mentioned that it does not support tail call optimization. In fact, many Lisps don't. As Bill Six pointed out even the ANSI standard of Common Lisp standard does not mandate (unlike Scheme) tail-call elimination provided by the language implementation, although it seems that all well-known ANSI Common Lisp compilers do it anyway.

I was wondering if there is a way to circumvent this problem and the first solution I found was to use memoize macro described in an excellent online documentation for newLISP, Code Patterns in newLISP:

(define-macro (memoize mem-func func)
  (set (sym mem-func mem-func)
    (letex (f func  c mem-func)
      (lambda ()
        (or (context c (string (args)))
        (context c (string (args)) (apply f (args))))))))

You can apply this macro to any function with any number of arguments. The trick here is that each time a function is called its result is cached in memory for another call. It can speed up your application tremendously, which can be observed by comparing the execution time of these example Fibonacci functions:

(define (fibo n)
  (if (< n 2) 1
    (+ (fibo (- n 1)) (fibo (- n 2)))))

(memoize fibo-m
  (lambda (n)
    (if (< n 2) 1
      (+ (fibo-m (- n 1)) (fibo-m (- n 2))))))

(time (fibo 35))
(time (fibo-m 35))

On my laptop (fibo 35) took 12,98 seconds to execute, while (fibo-m) executed in 0,016 miliseconds.

Unfortunately the memoize macro cannot help you to handle mutual recursion. A classic example of such recursion looks as follows:

(define (f1 n)
  (println n)
    (if (= n 0)
      (println "Blastoff!")
      (f2 n)))

(define (f2 n)
  (f1 (- n 1)))

You can easily make newLISP run out of stack by running (f1 1000), not to mention bigger numbers. What happens if we define a "memoized" version of f1 and f2? Let's see:

(memoize f1
  (lambda (n)
    (println n)
    (if (= n 0)
      (println "Blastoff!")
      (f2 n))))

(memoize f2
  (lambda (n)
    (f1 (- n 1))))

Again, running (f1 1000) immediately exhausts newLISP's stack.

A solution to this problem is using a technique called trampolining. Bill Clementson on his blog not only explained in an excellent way the concept of using trampolines, but also provided a Common Lisp implementation, which became my inspiration to write a newLISP version:

(define (trampoline fun arg)
  (catch
    (while true
      (let ((run (apply fun arg)))
        (setf fun (first run) arg (rest run)))) 'result)
  result)

A trampoline iteratively executes code thunks returned by a function, and this way avoids blowing out application stack. However, in order to use trampoline, the function must return continuation (a pointer to the next step) instead of value. Below is a version of beforementioned functions modified to use the trampoline:

(define (f1 n)
  (println n)
  (if (= n 0)
    (throw "Blastoff!")
    (list f2 n)))

(define (f2 n)
  (list f1 (- n 1)))

Now you can test it with:

(trampoline f1 '(1000))
(trampoline f1 '(10000))
(trampoline f1 '(100000))
...

Have fun!

MapReduce in Erlang

MapReduce is a Java framework for parallel computations, designed and developed by Google. It is often referred to not only as a platform, but also more general as an algorithm (or a "programming model", as Google calls it itself). The main idea behind MapReduce is to spawn computations on a set of data to many single processes (map), and then gather their results (reduce).

Implementing MapReduce in Erlang is unbelievably trivial. Joe Armstrong, Erlang inventor and main architect, published a sample pmap/2 function, which can be used as a parallel equivalent of standard Erlang lists:map/2:

pmap(F, L) ->
    S = self(),
    Pids = lists:map(fun(I) ->
                         spawn(fun() -> do_f(S, F, I) end)
                     end, L),
    gather(Pids).

gather([H|T]) ->
    receive
        {H, Ret} -> [Ret|gather(T)]
    end;
gather([]) ->
    [].

do_f(Parent, F, I) ->
    Parent ! {self(), (catch F(I))}.

Joe didn't provide any particular examples of using pmap, so I wrote a test function that makes a sequence of Fibonacci numbers. First, let's start an Erlang shell with -smp flag, that enables it to use many CPUs and many CPU cores:

erl -smp

Now define a Fibonacci function:

fib(0) -> 0;
fib(1) -> 1;
fib(N) when N > 0 -> fib(N-1) + fib(N-2).

and generate a test list as a sequence of numbers (for example from 0 to 40):

L = lists:seq(0,40).

To make every element of the list L to be processed by fib/1 function we use lists:map/2:

lists:map(fun(X) -> fib(X) end, L).

To do a parallel computation we need to use pmap/2 instead of lists:map/2:

pmap(fun(X) -> fib(X) end, L).

On my laptop with Core2 Duo T5450 1.66 Ghz running the parallel version of fib/1 function reduced execution time from 53.405534 to 31.354125 seconds (as measured by timer:tc/3). On dual Xeon E5430 2.66 Ghz (8 cores total) the speedup was from 25.635218 seconds to 10.017358. It is less than Joe Armstrong managed to achieve (he probably used more sophisticated test functions and different hardware than a regular PC), but it clearly shows how easy it is to parallelize an Erlang program - just replace every instance of map/2 with pmap/2 and you're done.

I prepared a modified version of pmap/2, which spawns processes not only on local machine, but also on all nodes in a cluster:

pmap(F, L) ->
    S = self(),
    Nod = [node()|nodes()],
    {Pids, _} = lists:mapfoldl(
        fun(I, {N1, N2}) ->
            case N1 == [] of
                true -> N = N2;
                false -> N = N1
            end,
            [H|T] = N,
            {spawn(H, fun() -> do_f(S, F, I) end), {T, N2}}
        end, {Nod, Nod}, L),
    gather(Pids).

Using the modified pmap/2, I got similar results with a cluster of two Erlang shells running on a single Core2 Duo machine as with Joe's function on a single shell started in smp mode. This version of pmap/2 also allows you to rebalance computations by starting as many Erlang shells as there are CPUs (or CPU cores) on multiple machines, and joining them in one cluster.