Concurrent Basic – A Declarative Language for Message-Based Concurrency.
Concurrent Basic represents a possible future for Visual Basic. Though based on work done in C# research languages such as Polyphonic C# and C-Omega, Visual Basic was chosen for its inherent predisposition towards declarative programming. The syntax is even inspired by VB’s declarative event handlers.
- Asynchronous – A method that represents a message queue.
- Synchronous – A method that represents a request queue.
- When – Used to wire Asynchronous and Synchronous events to each other
Below is a complete example of a thread-safe buffer implemented using these new keywords.
Class Buffer(Of T) Asynchronous Sub Put (t as T) Synchronous Function Take() as T Function React (t as T) as T When Put, Take Return T End Function End Class
The function Put can by called asynchronously by consumers. The Take function will block until there is at least one pending Put. The React function handles the implementation details of converting the object queued up via Put to the form expected by Take.
Below are two more examples. The first waits for a message on either A or B. The second example waits until there is a message on both A and B.
Class Either(Of T) Asynchronous Sub PutA(t as T) Asynchronous Sub PutB(t as T) Synchronous Function TakeEither() as T Function ReactA(t as T) as T When TakeEither, PutA Return T End Function Function ReactB(t as T) as T When TakeEither, PutB Return T End Function End Class Class Both(Of T, U) Asynchronous Sub PutA(t as T) Asynchronous Sub PutB(u of U) Synchronous Function TakeBoth() as T Function React(t as T, u as U) as Pair(Of T, U) When Take, PutA, PutB Return new Pair(Of T, U)(t, u) End Function End Class
Next up is a pattern for a one place buffer, which is a buffer that stores at most one message at a time.
Class OPB(of T) Sub New Empty() End Sub Synchronous Sub Put (t as T) Synchronous Function Take() as T Private Asynchronous Empty() Private Asynchronous Full(of T) Sub PutAndEmpty(t as T) When Put, Empty Full(t) End Sub Sub TakeAndFull(t as T) as T When Take, Full Empty() Return t End Sub End Class
Under the hood, Put, Take, Empty, and Full represent internal queues. The reaction methods are executed via pattern matching between the When clauses and the size of the queues.
Up until now each example assumed that the Take method would be synchronous. It doesn’t have to be however, it can also use a callback.
Class AsyncBuffer(Of T) Asynchronous Sub Put (t as T) Asynchronous Function Take() as T Function React (callback as Func(t as T), t as T) as T When Put, Take Return callback(t) End Function End Class
Using this model, a thread is spawned to process the React and callback functions. Some people may want to do something else, such as use a thread pool, GUI thread, or some other threading library. To do this you need to implement a ContinuationAttribute. Below is the base-class for the attribute.
Public MustInherit Class ContinuationAttribute Inherits Attribute Public MustOverride Sub BeginInvoke( task As Continuation) Public MustOverride Sub Invoke( task As Continuation) End Class Public Delegate Sub Continuation()
Example and usage
Public Class MyThreadPoolAttribute Inherits ContinuationAttribute Public Overrides Sub BeginInvoke( task As Continuation) ThreadPool.Enqueue(task) End Sub Public Overrides Sub Invoke( task As Continuation) task() End Sub End Class
Class AsyncBuffer(Of T) Asynchronous Sub Put (t as T) Asynchronous Function Take() as T
Function React (callback as Func(t as T), t as T) as T When Put, Take Return callback(t) End Function End Class