Commit c31a3371 authored by Francois-Rene Rideau's avatar Francois-Rene Rideau

Import from mtn repo

LISP=sbcl setup.lisp
cl-launch -p ${LISP_SYSTEMS} -p ${FARMER_PATH} -l ${LISP} -f $< -s ${SYSTEM} -r philip-jose::farmer-start -o $@
PHILIP-JOSE, a farmer for distributed computations in Lisp
This package implements a "farmer": a program to control a farm of computers
and coordinate them into achieving a large computation.
The basic usage model is that all the flow control happens in the farmer,
whereas all the intensive computation happens in worker processes
each of which needs only know how to do a simple part of the whole.
The overall computation may have many components and sub-components
that include both massive parallelism and sequential dependencies,
where the control-flow that may depend on the intermediary results,
where failure of components can happen, be detected and acted upon, etc.
This software is released under the bugroff license. Use at your own risk.
At the insistence of several hackers, I hereby state what is obvious to me,
that they can reuse any software released under the bugroff license and
publish it as part or totality of packages under any other license they see fit
if it really matters to them, including a BSD-style license or a MIT license.
Yes they can. Of course, if they choose a proprietary software license,
they only deserve scorn. But even that, they may do!
Communication happens through a simple request/response protocol (like HTTP).
The current protocol is designed to make it very easy to pass around
Lisp data as the arguments and results of computations. It does that well,
but its current incarnation has many limitations. Most importantly,
the current system is intrinsically unsecure. DO NOT USE IT ON THE INTERNET.
Use it only within a trusted network, or wrapped over encrypted lines.
Also, the protocol is NOT designed for transmitting large amounts of data.
If workers need share data with the farmer and/or with each other,
they should access it through independent means, such as a file server.
The farmer as it currently is is only designed to handle
the general flow control of the computation.
Workers are spawned by the farmer itself, using ssh or fork+exec,
from a farm of registered machines.
A simple mechanism to register and unregister machines is provided.
A validation may be run on each worker machine
to qualify it or not for the current computation.
Because simple forking (without exec) is not (currently) supported
(and because it is not usually available for distributed computations),
an independent means of storing and retrieving data
is once again needed if workers are to share it.
Flow Control
On the plus side, the flow-control of the computation
can be written in a very natural way.
The farmer supports for sequential, parallel and first-wins subcomputations.
Moreover, it provides a universal primitive (i.e. delimited continuations)
on top of of which you can build your own control structures
if you ever need something more elaborate.
And indeed, the previous high-level control structures
were built in terms of these primitives.
I am told the implementation of continuations I am using (arnesi)
allows continuations to be serialized, so in theory,
you could build upon this to achieve mobile code between several farmers
(as in the Tube), persistent threads, load-balancing, etc.
Note that arnesi's call/cc and kall are more like shift and reset
than like Scheme's call/cc. I admit I haven't tried nesting them
syntactically and see if it still does the right thing, and even less to
see if it handles exceptions properly.
I'd have to compare the results to those of the examples given
in the following documents:
The farmer is written with the assumption that it will work
on a sequential Lisp implementation, or
in a single thread of a multithreaded Lisp implementation.
Its concurrent programming activities are built out of
a home-grown green thread mechanism on top of arnesi's call/cc.
Philip-jose provides green threads intended to be used on the farmer.
Because of their execution model, all computations between two calls
to some threading or I/O primitives are done in a same atomic transaction.
The upside is that you can get a lot done without ever having to use locks
or any other explicit mutual exclusion mechanism --
just avoid calling any computation with complex control flow between two
operations the results of which should be seen atomically by other threads.
Many places in the existing code indeed assume this sequential execution
to access to various shared data-structures in atomic transactions
without the explicit use of locks.
The downside is that when you really need such mutual exclusion mechanism,
it is not provided (yet), and other parts of the code haven't been made
to use any such mechanism (yet). Writing and providing such mechanisms
should be pretty easy however, and are let as an exercise to the astute
Though a skeleton of multithreaded server is provided,
it is not tested, probably buggy, lacking in features, and
will not work in conjunction with other code in philip-jose.
I'm told that the author of IOLib is building
a better, more featureful, more stable, infrastructure for I/O in Lisp.
Philip-Jose is no such infrastructure.
TO-DO List
1- package the system into something usable. Provide documentation, etc.
2- use non-blocking IO to provide for better networking in farmers:
the ability to be both a server and client, etc.
3- find how to serialize continuations to provide richer capabilities:
persistent (more robust) threads, and mobile code (including load balancing).
4- Implement fork() as a more efficient (though trickier) alternative to
fork+exec() for spawning new clients that share initial state.
5- provide a faster alternative to arnesi's with-call/cc, based on e.g. Screamer.
6- implement interfaces that are compatible with existing distributed systems?
(Scheme) Termite, Askemos, Tube, Kali, Dreme, [Something by Queinnec]
(CL) StarLisp, NetCLOS, GBBopen, ...
7- use hunchentoot or such to support HTTP instead of the current protocol.
8- implement a simple map/reduce for local-tasks, for worker-jobs
9- implement pure streams, and mappings between pure & impure streams,
so that I/O in competing threads only gets committed in the winning thread;
also allows for easy backtracking with I/O operations. One thread may
(read-string "foo" s) and just catch the exception and die if not available.
10- have a get-read-buffer interface to not read character-by-character
11- have thread IDs, and guard various event-handlers with a test for the thread
still being alive. Alternatively have a mechanism to (atomically) remove event-handlers,
queued jobs, etc., from their respective queue, set, etc.
12- enforce atomicity with a global flag \*atomic* or such; when this flag is on,
weak yields turn into NOPs and strong yields raise error when it is on.
Now we also have to distinguish between weak yields and strong yields.
13- Have sub-queues, etc., for scheduling.
14- provide some explicit mechanisms for mutual-exclusion and transactionality.
15- merge with Erlang-in-Lisp
Behaving like an Erlang node
* see distel:
* the Erlang distribution protocol can be found in the source release at
Using ancillary data to BSD UNIX sockets
If you are send multiple fds in a same sendmsg()
make sure you send them all in a single CMSG
because multiple CMSGs of same type are broken on many kernels.
man cmsg. See SCM_RIGHTS.
;;; -*- Mode: Lisp ; Base: 10 ; Syntax: ANSI-Common-Lisp -*-
;;;;; Main loop for the Philip-Jose Farmer
;;; Manage a farm of processes each doing a part of a grand job.
(in-package :philip-jose)
(defparameter *philip-jose-farmer-version* 0)
(defun-request-handler :farmer-version ()
(reply *philip-jose-farmer-version*))
(defun-request-handler :square (n)
(reply (* n n)))
(defun-request-handler :1+ (n)
(reply (1+ n)))
(defun-request-handler :reload ()
(reply (asdf:oos 'asdf:load-op *farmer-system*)))
(defun-request-handler :register-client (id &key valid explanation)
(logger "~&Registering client~{ ~A~}" id)
(reply t))
(defun-request-handler :armageddon ()
(flet ((kill-worker-id (worker-id)
(destructuring-bind (machine process connect-time) worker-id
(declare (ignorable connect-time))
(logger "~&Armageddon killing pid ~D on ~A" process machine)
(kill-machine-process machine process))))
(maphash (lambda (worker-id status)
(declare (ignorable status))
(kill-worker-id worker-id))
(maphash (lambda (job-id job)
(declare (ignorable job-id))
(kill-worker-id (car (job-status job))))
(reply t)
(logger "~&Armageddon quitting")
(defun-job fake (m)
(setf *random-state* (make-random-state t))
(let ((x (* 1d-3 (random 10000))))
(DBG :fake m x)
(fsleep x)
(values m x)))
(defun show-tasks ()
(let ((*print-level* 4))
(logger "~&scheduled local-tasks: ~S~%" (fifo-head *scheduled-local-tasks*))
(logger "~&timed local-tasks: ~S~%" (container-contents *timed-local-tasks*))
(logger "~&scheduled worker jobs: ~S~%" (fifo-head *scheduled-worker-jobs*))
(logger "~&claimed worker jobs: ~S~%" (hash-table->alist *claimed-worker-jobs*)))
(defun clear-tasks ()
(fifo-empty! *scheduled-local-tasks*)
(empty-container! *timed-local-tasks*)
(fifo-empty! *scheduled-worker-jobs*)
(clrhash *claimed-worker-jobs*)
(trace call-request-handler handle-sexp-request)
(hash-table->alist *request-heads*)
;;; -*- Mode: Lisp ; Base: 10 ; Syntax: ANSI-Common-Lisp -*-
;;;;; Forking
(in-package :philip-jose)
(defvar *expected-children 2.1)
(defvar *expected-space* 16000000))
;; Simple heuristic: if we have allocated more than the given ratio
;; of what is allowed between GCs, then trigger the GC.
;; note: can possibly modify parameters and reset in sb-ext:*after-gc-hooks*
(defparameter *prefork-allocation-reserve-ratio* .10) ; default ratio: 10%
(defun should-i-gc-p ()
(let ((available-bytes (- (sb-alien:extern-alien "auto_gc_trigger" sb-alien:long)
(allocation-threshhold (sb-ext:bytes-consed-between-gcs)))
(< available-bytes (* *prefork-allocation-reserve-ratio* allocation-threshhold)))))
(defun pre-fork-hook ()
(when (should-i-gc-p)
(defun do-fork ()
#-sbcl (et:fork)
#+sbcl (sb-posix:fork))
(defun post-fork-child-cleanup ()
(defun post-fork-parent-hook (pid)
(defun fork ()
(let ((pid (do-fork)))
(if (zerop pid)
(post-fork-parent-hook pid))
(defvar *cleanup-pusher* nil
"dynamically-scoped function for declaring cleanups")
(defun push-cleanup (cleanup)
(if *cleanup-pusher*
(funcall *cleanup-pusher* cleanup)
(error "No cleanup pusher declared")))
(defun call-with-cleanups (thunk)
(let ((cleanups nil))
(flet ((push-cleanup (cleanup) (push cleanup cleanups)))
(funcall thunk #'push-cleanup)
(dolist (cleanup cleanups)
(funcall cleanup))))))
(defun call-with-foreign-temporaries (thunk)
(let (temporaries)
(flet ((alloc (s)
(push s temporaries)))
(funcall thunk alloc)
(loop for s in temporaries do
(apply #'cffi:free-converted-object s)))))))
(defmacro with-foreign-temporaries (&body body)
(with-gensyms (s)
(lambda (,s)
(macrolet ()
(cffi:defcenum frobork-kode
(cffi:defcfun (c-frobork "frobork") :int (kode :pointer))
(defun frobork-convert (v)
(typecase v
(cffi:convert-to-foreign v :string))
(defun frobork (kl)
(let ((l (length kl)))
(with-foreign-object (kode :int (1+ l))
(let ((alloc-state
(loop with converted = nil
with state = nil
for i below l
for v in kl
do (multiple-value-setq (converted state)
(frobork-convert v))
(setf (mem-aref kode :int i) converted)
when state
collect state)))
(setf (mem-aref kode :int l) 0)
(c-frobork kode)
(loop for s in alloc-state
do (apply #'cffi:free-converted-object s)))))))
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
enum frobork_kode {
* frobork.c
#define XX (*kode++)
#define T(type) ((type)XX)
#define X T(int)
#define TRY2(x,y) if((x)==(y)) { goto on_error; } ; break;
#define TRY(x) TRY2(x,-1)
#define OBJLEN(x) (x),sizeof(x)
#define GETX {x=X;}
#if 0 /* support for any of the types we use being larger than an int */
#define MAXUINTP1 (1ULL+(unsigned long long)((unsigned int)-1))
#define SIZUI (sizeof(unsigned int))
#define SIZULL (sizeof(unsigned int))
unsigned long long read_and_increment_pointer (unsigned int**p, unsigned int size) {
unsigned long long r;
unsigned int *q = *p;
unsigned int i = 0;
unsigned long long factor = 1;
if (size <= SIZUI) {
r = *q;
} else if (size > SIZULL) {
write(2,OBJLEN("frobork: bad int size\n"));
} else {
r = 0;
while (i<size) {
r += *q*f;
*p = q;
return r;
#define T(type) ((type)read_and_increment_pointer((unsigned int**)&kode,sizeof(type)))
int frobork (int* kode)
pid_t pid, p1,p2;
int x;
sighandler_t sh;
char *a,**b,**c;
pid = fork();
if (pid) {
return pid;
while (1) {
switch (X) {
a=T(char *);b=T(char **);c=T(char **);
case FK_DUP2:
GETX; TRY(dup2(x,X));
case FK_CLOSE:
switch(X) {
case 0: sh = SIG_...;
case 1: sh = SIG_...;
case 2: sh = SIG_...;
write(2,OBJLEN("frobork: bad code\n"));
return 0;
int main (int argc, char** argv) {
int kode[] = {
FK_DUP2, 0, 4,
printf("pid=%d frobork returned %d\n",getpid(),frobork(kode));
return 0;
;;; -*- Mode: Lisp ; Base: 10 ; Syntax: ANSI-Common-Lisp -*-
;;;;; Main loop for the Philip-Jose Farmer
;;;;; Incremental parser protocol
(in-package :philip-jose)
;;;; The Incremental parser protocol
;; This protocol allows for a single-threaded server to monitor many connections at once,
;; possibly on many ports, with each connection having partial request message that will
;; either be trivially handled, or lead to spawning after the request is complete.
;; An incremental parser function is given a buffer (byte or char array),
;; the size up to which the buffer is filled, and
;; the previous (incomplete) state of the incremental parser, or nil on the first attempt.
;; It returns two values. The first is a boolean that tells if the parse was successful.
;; If the first value is T, then the second value is the result of the parse.
;; If the first value is NIL, then the second value is the state of the incremental parser,
;; from which to resume when the buffer is made more full.
;; In a world with any kind of first-class (partial) continuations, either linear or multi-entry,
;; any direct parser could be made incremental, by having the READ-CHAR primitive return the
;; partial continuation of the parser to the caller.
;; Maybe we can hook Screamer or arnesi's call/cc into here.
;; TODO:
;; * add a buffer management protocol to make the thing work for bigger requests and
;; on-going conversations
;;;; Trivial implementations of simple cases
;;; Line parser -- the request in complete at the end of the first line
(defun line-end-code-p (cc)
(member cc '(10 13)))
(defun line-end-char-p (ch)
(line-end-code-p (char-code ch)))
(defun trivial-incremental-line-parser (buffer size state)
;; Implements the incremental parser protocol for character input, detecting a full line.
(let* ((line-end-pos (position-if #'line-end-char-p buffer :start (or state 0))))
(if line-end-pos
(values t line-end-pos)
(values nil size))))
;;; Wrapper that prevents too many retries
;; Drop (presumably buggy or malevolent) connections with too many packets for the initial request.
;; As in attempts to flood with small packets, connection with a lot of fragmentation,
;; malformed request, overlarge request, buffer overflow tentative.
(define-condition too-many-attempts ()
(:report (lambda (condition stream)
(declare (ignore condition))
(format stream "Too many attempts in incremental request parsing"))))
(defun incremental-parser-attempt-limiter (max-attempts incremental-parser)
(labels ((check (x)
(when (>= x max-attempts)
(error 'too-many-attempts)))
(try (buffer size state)
(multiple-value-bind (count inner-state)
(if state (values (car state) (cdr state)) (values 0 nil))
(check count)
(multiple-value-bind (successp x)
(funcall incremental-parser buffer size inner-state)
(if successp
(values t x)
(let ((c (1+ count)))
(check c)
(values nil (cons c x))))))))
(if (typep max-attempts '(unsigned-byte 16))
;;; Trivial transformation of a parser into a pseudo incremental parser.
;; Restarting from the beginning everytime.
;; It's inefficient and a big DoS target unless combined with the limiter above.
(defun pseudo-incremental-parser-from-reader (reader)
;; Implements the incremental parser protocol for character input, given a reader.
;; Not really incremental, always restart from scratch, so called pseudo.
;; Could do better with call/cc and an open stream protocol.
#'(lambda (buffer size state &aux index)
(declare (ignore state))
(values t (cons (with-input-from-string (s buffer :index index :start 0 :end size)
(funcall reader s t))
(end-of-file () (values nil nil)))))
;;; Use of the Lisp reader as a parser.
;; NOT FOR USE IN THE WILD! See warning in trivial-sexp-server.lisp
(let ((f (pseudo-incremental-parser-from-reader #'safe-read)))
(defun make-incremental-reader (&optional (max-attempts 4))
(incremental-parser-attempt-limiter max-attempts f)))
(defun incrementally-parsed-request-handler (connection buffer size state &optional (handler #'handle-sexp-request))
(with-trivial-logging ()
(destructuring-bind (request . index) state
(with-open-stream (bufin (make-string-input-stream buffer index size))
(with-open-stream (in (make-concatenated-stream bufin connection))
(let ((*standard-input* in)
(*standard-output* connection))
(funcall handler request)))))))
(defun make-incrementally-parsed-request-handler (h)
#'(lambda (connection buffer size state)
(incrementally-parsed-request-handler connection buffer size state h)))
;;; -*- Mode: Lisp ; Base: 10 ; Syntax: ANSI-Common-Lisp -*-
;;;; Manage tasks in the local Lisp image
;;;; Tasks can be scheduled on a queue,
;;;; to be processed either immediately or when a given start time arrives.
(in-package :philip-jose)
;;(declaim (optimize (speed 3) (safety 3) (debug 3)));DEBUG
;; All of this assumes single-threaded execution.
;; If you want concurrent execution, you need to either modify this code and add locking,
;; Or to isolate instances that run this code in separate threads that each have a
;; thread-local binding for each of the special variables used.
;;; Tasks
(defclass local-task ()
((callback :accessor task-callback :initarg :callback)
(name :accessor task-name :initarg :name :initform nil)))
(defclass timed-local-task (local-task)
((start-time :accessor task-start-time :initarg :start-time :initform nil)))
(defvar *current-local-task* nil)
;;; Scheduled tasks (to executed ASAP in FIFO order)
(defparameter *scheduled-local-tasks*
"tasks that need to be executed locally")
(defun enqueue-immediate-local-task (task)
(fifo-enqueue task *scheduled-local-tasks*))
(defun schedule-immediate-local-task (callback &key name)
(make-instance 'local-task
:callback callback :name name))
(defun process-local-tasks ()
(loop until (fifo-empty-p *scheduled-local-tasks*) do
(let ((*current-local-task* (fifo-dequeue *scheduled-local-tasks*)))
(funkall (task-callback *current-local-task*)))))
;;; Timed tasks
(defun time-sooner-p (t1 t2)
((null t2) nil)
((null t1) t)
(t (< t1 t2))))
(defun task-scheduled-sooner-p (task1 task2)
(time-sooner-p (task-start-time task1) (task-start-time task2)))
(defparameter *timed-local-tasks*
(make-instance 'binary-heap :lessp #'task-scheduled-sooner-p)
"tasks to be executed on farmer at a given time. e.g. timeout functions")
(defun schedule-timed-local-task (callback &key start-time name)
(make-instance 'timed-local-task
:callback callback :start-time start-time :name name))
(defun next-timed-task-start-time ()
(when-bind task
(unless (container-empty-p *timed-local-tasks*)
(least-item *timed-local-tasks*))
(task-start-time task)))
(defun process-timed-task-heap (&optional (time (et:gettimeofday)))
(loop for next-start-time = (next-timed-task-start-time)
while (and next-start-time (< next-start-time time)) do
(enqueue-immediate-local-task (pop-least-item! *timed-local-tasks*))))
(defun schedule-local-task (callback &key start-time name)
(if start-time
(schedule-timed-local-task callback :start-time start-time :name name)
(schedule-immediate-local-task callback :name name))
;;; Event-Base
(defun process-events ()
(if (iomux::event-base-empty-p *event-base*)
;;(defparameter *event-processing-task*
;; (make-instance 'local-task
;; :name "Event processor"
;; :callback #'process-events))
;;(defun reschedule-process-events ()
;; (schedule-local-task *event-processing-task*))
(defun process-no-events ()
(if-bind time (next-timed-task-start-time)
(let ((delay (- time (get-real-time))))
(fsleep delay)