Asynchronous API

Use the Aerospike C# client library asynchronous API (AsyncClient) to queue commands and return control to the application while the commands are processed with non-blocking sockets.

The AsyncClient instance is thread-safe and can be used concurrently. It does not dedicate a thread to each command: commands are handed to a scheduler that owns a fixed set of command slots and drives non-blocking sockets, so a command in flight occupies a connection rather than a thread. In addition to single-record commands, the asynchronous surface includes batch, scan, query, UDF execution, and transaction operations.

Concurrency limits

asyncMaxCommands limits active asynchronous commands. asyncMaxCommandAction controls what happens at that limit: REJECT throws immediately, BLOCK waits for a slot, and DELAY queues the command. When using DELAY, asyncMaxCommandsInQueue bounds that queue. Exceeding it rejects the command.

AsyncClientPolicy policy = new()
{
  user = args.user,
  password = args.password,
  clusterName = args.clusterName,
  tlsPolicy = args.tlsPolicy,
  authMode = args.authMode,
  asyncMaxCommands = args.commandMax,
  asyncMaxCommandAction = MaxCommandAction.DELAY,
  asyncMaxCommandsInQueue = 1000,
  useServicesAlternate = args.useServicesAlternate,
  failIfNotConnected = true
};

Choosing an API

AsyncClient serves high concurrency with fewer threads than the synchronous client, and that is the reason to reach for it. It is worth the added complexity when you need many commands in flight and cannot afford a thread parked on each one. When a bounded thread pool already handles your throughput, AerospikeClient is easier to write, debug, and operate, and is the better default.

Note

AsyncClient is built on its own command scheduler and non-blocking sockets rather than on the Task Parallel Library, and the Task overloads were added later as an adapter over the callback API. Because the client schedules and completes commands itself, it does not scale or behave the way a TPL-based library does, and sizing it means tuning the policies below rather than the .NET thread pool.

Task overloads and listener callbacks

The two surfaces are alternative front ends to the same machinery, not two implementations. The Task overloads create a listener internally and complete a TaskCompletionSource from it. They give you async/await syntax over the client’s own scheduling, which means a few behaviors differ from what that syntax normally implies:

• Concurrency is capped by asyncMaxCommands, not by the thread pool.

  Awaiting more commands than the limit allows does not queue them in the TPL. It triggers asyncMaxCommandAction.

• The default BLOCK action blocks the calling thread.

  At the limit, the call blocks until a slot frees up, and it does so before returning a task, so await does not keep the caller’s thread free. Choose DELAY or REJECT if callers must stay unblocked.

• A rejected command throws from the call rather than faulting the task.

  AerospikeException.CommandRejected is raised while the command is being scheduled. Awaiting the call directly still catches it, but storing the task and awaiting it later does not.

• Continuations can run on the thread that completed the command.

  Work placed immediately after an await may occupy a thread the client needs to process other commands, so keep it short and move blocking or long-running work off that path.

Neither surface is deprecated. Use the Task overloads for ordinary control flow, or the listener callbacks to avoid a task allocation per command on hot paths.

Example

Listener callbacks:

client.Put(policy, new WriteHandler(this, key, bin), key, bin);
WaitTillComplete();

The equivalent Task overloads:

using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
CancellationToken token = cancellation.Token;
await client.Put(writePolicy, token, key, bin);

Record record = await client.Get(policy, token, key);

Cancellation cancels the client-side Task, but it does not guarantee that an already submitted database command is stopped. A write can therefore complete on the server after the waiting task is canceled. Use idempotent operations or verify state when that distinction matters.

The complete listener example below demonstrates connecting, writing, reading, and waiting for callbacks before closing the client.

using System;
using System.Threading;
using Aerospike.Client;

namespace Test
{
    public class AsyncTest
    {
        private AsyncClient client;
        private WritePolicy policy;
        private bool completed;

        public AsyncTest()
        {
            policy = new WritePolicy();
        }

        public void RunTest()
        {
            client = new AsyncClient("127.0.0.1", 3000);
            try
            {
                // Write a single value.
                Key key = new Key("test", "myset", "mykey");
                Bin bin = new Bin("mybin", "myvalue");
                Console.WriteLine(string.Format("Write: namespace={0} set={1} key={2} value={3}",
                key.ns, key.setName, key.userKey, bin.value));
                client.Put(policy, new WriteHandler(this, key, bin), key, bin);
                WaitTillComplete();
            }
            finally
            {
                client.Close();
            }
        }

        private class WriteHandler : WriteListener
        {
            private readonly AsyncTest parent;
            private readonly Key key;
            private readonly Bin bin;

            public WriteHandler(AsyncTest parent, Key key, Bin bin)
            {
                this.parent = parent;
                this.key = key;
                this.bin = bin;
            }

            public void OnSuccess(Key key)
            {
                try
                {
                    // Write succeeded.  Now call read.
                    parent.client.Get(parent.policy, new RecordHandler(parent, key), key);
                }
                catch (Exception e)
                {
                    Console.WriteLine(string.Format("Failed to get: namespace={0} set={1} key={2} exception={3}",
                    key.ns, key.setName, key.userKey, e.Message));
                }
            }

            public void OnFailure(AerospikeException e)
            {
                Console.WriteLine("Failed to put: namespace={0} set={1} key={2} exception={3}",
                key.ns, key.setName, key.userKey, e.Message);
                parent.NotifyCompleted();
            }
        }

        private class RecordHandler : RecordListener
        {
            private readonly AsyncTest parent;
            private readonly Key key;

            public RecordHandler(AsyncTest parent, Key key)
            {
                this.parent = parent;
                this.key = key;
            }

            public void OnSuccess(Key key, Record record)
            {
                // Read completed.
                object received = (record == null) ? null : record.GetValue("mybin");
                Console.WriteLine(string.Format("Received: namespace={0} set={1} key={2} value={3}",
                key.ns, key.setName, key.userKey, received));
                // Notify application that read is complete.
                parent.NotifyCompleted();
            }

            public void OnFailure(AerospikeException e)
            {
                Console.WriteLine("Failed to get: namespace={0} set={1} key={2} exception={3}",
                key.ns, key.setName, key.userKey, e.Message);
                parent.NotifyCompleted();
            }
        }

        private void WaitTillComplete()
        {
            lock (this)
            {
                while (!completed)
                {
                    Monitor.Wait(this);
                }
            }
        }

        private void NotifyCompleted()
        {
            lock (this)
            {
                completed = true;
                Monitor.Pulse(this);
            }
        }
    }
}

Application Wait

To make the application wait and avoid closing the connection before the write and read commands complete:

private void WaitTillComplete()
{
    lock (this)
    {
        while (!completed)
        {
            Monitor.Wait(this);
        }
    }
}

private void NotifyCompleted()
{
    lock (this)
    {
        completed = true;
        Monitor.Pulse(this);
    }
}