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.

Keywords

• 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

For more information you can watch the Channel 9 video and the Concurrent Basic proposal.