{ "cells": [ { "cell_type": "markdown", "metadata": { "toc": true }, "source": [ "

Table of Contents

\n", "
" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Understanding Asynchronous Operations in Aerospike\n", "This tutorial describes asynchronous operations in Aerospike: why they are used, the architecture, and how to program with async operations. \n", "\n", "This notebook requires the Aerospike Database running locally with Java kernel and Aerospike Java Client. To create a Docker container that satisfies the requirements and holds a copy of Aerospike notebooks, visit the [Aerospike Notebooks Repo](https://github.com/aerospike-examples/interactive-notebooks)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Introduction\n", "In this notebook, we will see the benefits, design, and specifics of programming with asynchronous operations in Aerospike. \n", "\n", "Aerospike provides asynchronous APIs for many operations. We will describe the benefits of using async operations and key abstractions in the client related to async requests. After covering the theoretical ground, we will show how it all comes together with specific code examples. \n", "\n", "The notebook tutorial has two parts: \n", "- architecture and concepts, and \n", "- coding examples.\n", "\n", "The main topics include:\n", "- Execution models in Aerospike\n", "- Benefits of async\n", "- Key concepts\n", "- Framework for async programming\n", "- Coding examples" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Prerequisites\n", "This tutorial assumes familiarity with the following topics:\n", "- [Aerospike Notebooks - Readme and Tips](../readme_tips.ipynb)\n", "- [Hello World](hello_world.ipynb)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Initialization" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Ensure database is running\n", "This notebook requires that Aerospike Database is running. " ] }, { "cell_type": "code", "execution_count": 1, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:49.065421Z", "start_time": "2020-12-29T20:48:49.060897Z" } }, "outputs": [], "source": [ "import io.github.spencerpark.ijava.IJava;\n", "import io.github.spencerpark.jupyter.kernel.magic.common.Shell;\n", "IJava.getKernelInstance().getMagics().registerMagics(Shell.class);\n", "%sh asd" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Download and install additional components.\n", "Install the Aerospike Java client and the Java Netty package, which is described later in the notebook." ] }, { "cell_type": "code", "execution_count": 2, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:50.084636Z", "start_time": "2020-12-29T20:48:50.080629Z" } }, "outputs": [], "source": [ "%%loadFromPOM\n", "\n", " \n", " com.aerospike\n", " aerospike-client\n", " 5.0.0\n", " \n", " \n", " io.netty\n", " netty-all\n", " 4.1.53.Final\n", " compile\n", " \n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Constants and Convenience Functions\n", "We will use some constants and convenience functions throughout this tutorial, including the namespace \"test\" and set \"async-ops\". " ] }, { "cell_type": "code", "execution_count": 3, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:50.771243Z", "start_time": "2020-12-29T20:48:50.767819Z" } }, "outputs": [], "source": [ "final String Namespace = \"test\";\n", "final String Set = \"async-ops\";\n", "\n", "// truncate data, close client and event loops - called multiple times to initialize with different options\n", "// described in greater detail later\n", "void Cleanup() {\n", " try {\n", " client.truncate(null, Namespace, Set, null);\n", " }\n", " catch (AerospikeException e) {\n", " // ignore\n", " }\n", " client.close();\n", " eventLoops.close();\n", "};" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Open a Terminal Tab\n", "You may execute shell commands including Aerospike tools like [aql](https://docs.aerospike.com/docs/tools/aql/index.html) and [asadm](https://docs.aerospike.com/docs/tools/asadm/index.html) in the terminal tab. Open a terminal tab by selecting File->Open from the notebook menu, and then New->Terminal." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Synchronous, Asynchronous, and Background Operations\n", "An application uses the Aerospike client library (aka the client) to interact with Aerospike Database. The client sets up a connection to the appropriate server node and sends in a request for execution in one of the following modes:\n", "\n", "Synchronous: The request thread makes the request, waits for the response, and processes the response upon arrival.\n", "\n", "Asynchronous: The request thread submits one or more requests, and results are processed in one or more callback thread(s) as they arrive.\n", "\n", "Background: The request thread submits the request and operation (or task) is completed in the background. The submission returns immediately, while the actual operation executes separately. The application can check the completion status of the task, and after it is completed may examine the results in the database with one or more separate requests. \n", "\n", "Note, a background operation may be considered a special type of asynchronous operation, and it is applicable only for updates in Aerospike. By asynchronous operations we refer to only those that return results in a callback." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Asynchronous Operations For Better Resource Efficiency\n", "During the time a request is sent to the server and the result arrives (“request latency”), the client and application need not wait idly if a high throughput is the goal. A higher throughput can be achieved through concurrent requests.\n", "\n", "- Synchronous: The application can spawn multiple threads and process multiple requests in parallel, one per thread at a time. \n", "- Asynchronous: The application can process requests asynchronously by submitting them in parallel without waiting for the results. The results are processed as they arrive in a different “callback” thread. An async request uses a dedicated connection to the server.\n", "- Pipeline: Multiple requests could be sent over the same connection to the same server node, and their results received over the same connection. Thus there is greater sharing of threads and connections across multiple requests. Aerospike currently does not support pipeline processing.\n", "\n", "In many cases, asynchronous processing can be more resource efficient and can deliver better throughput than multi-threaded synchronous processing because threads have memory and CPU (context-switch) overhead and their number may be limited by the OS. \n", "\n", "On the other hand, the asynchronous model is more complex to program and debug. The application should make judicious use of synchronous, asynchronous, and background requests. The client can perform different type of commands in a single instance. \n", "\n", "It should be noted that background operations when appropriate would typically deliver superior throughput especially in a non-UDF invocation." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Supported Asynchronous Operations\n", "Most CRUD operations have the async variant. \n", "\n", "- Single record operations\n", " - add, append, delete, apply(udf), get, getHeader, operate, prepend, put, touch\n", "- Batch operations: \n", " - exists (array listener and sequence listener), get (batch list and batch sequence listener), get (record array and record sequence listener), getHeader\n", "- Query/scan: Callback handles a series of records, a single record at a time.\n", " - query, queryPartitions\n", " - scanAll, scanPartitions\n", "- Metadata: createIndex, dropIndex\n", "- Operational: info\n", "\n", "Please refer to the [API documentation](https://docs.aerospike.com/apidocs/java/com/aerospike/client/AerospikeClient.html) for details.\n", "\n", "# Execution Model\n", "The async methods take two additional arguments than their sync variants: the “event loops” and “listener (callback)”. See the code in [Async Framework](#Async-Framework) section below.\n", "- Event loops: An event-loop represents the loop of \"submit a request\" and \"asynchronously process the result\" for concurrent processing of events or requests. Multiple event loops are used to leverage multiple CPU cores.\n", "- Listener: The listener encapsulates the processing of results. \n", " - Listener types: Depending on the expected number of records in the result and whether they arrive at once or individually, different listener types are to be used such as a single record listener, a record array listener, or a record sequence listener.\n", " - Completion handlers: A single record or record array is processed with the success or failure handler. In a record sequence listener, each record is processed with a \"record\" handler, whereas the success handler is called to mark the end of the sequence." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Application Call Sequence\n", "The application is responsible for spreading requests evenly across event loops as well as throttling the rate of requests if the request rate can exceed the client or server capacity. The call sequence involves these steps (see the code in [Async Framework](#Async-Framework) section below):\n", "- Initialize event loops.\n", "- Implement the listener with success and failure handlers. \n", "- Submit requests across event loops, throttling to stay below maximum outstanding requests limit.\n", "- Wait for all outstanding requests to finish." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Understanding Event Loops\n", "Let's look at the key concepts relating to event loops. As described above, an event loop represents concurrent submit-callback processing of requests. See the code in [Async Framework](#Async-Framework) section below.\n", "\n", "**Number of event loops**: In order to maximize parallelism of the client hardware, as many event loops are created as the number of cores dedicated for the Aerospike application. An event pool is aligned with a CPU core, not to a server node or a request type.\n", "\n", "**Concurrency level**: The maximum concurrency level in each event loop depends on the effective server throughput seen by the client, and in aggregate may not exceed it. A larger value would result in request timeouts and other failures.\n", "\n", "**Connection pools and event loops**: Connection pools are allocated on a per node basis, and are independent of event pools. When an async request needs to connect to a node, it uses a connection from the node’s connection pool only for the duration of the request and then releases it.\n", "\n", "**Connection pool size**: Concurrency across all loops must be supported by the number of connections in the connection pool. The connection pool per node should be set equal to or greater than the total number of outstanding requests across all event loops (because all requests may go to the same node in the extreme case). \n", "\n", "**Delay queue buffer**: To buffer a temporary mismatch in processing and submission rates, there is a delay queue buffer in front of an event loop where requests are held until an async request slot becomes available in the event loop. The queued request is automatically assigned to a slot and processed without involvement of the application. \n", "\n", "**Throttling**: The delay queue cannot buffer a long running mismatch in submission and processing speeds, however, and if the wait queue fills up, a request will not be accepted and the client will return “delay queue full” error. The application should throttle by keeping track of outstanding requests and issue a new request when an outstanding one finishes. If delay queue size is set to zero, throttling must also be handled in the application code." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Event Loop Variants: Netty, NIO, EPOLL\n", "Both Netty and Direct NIO event loops are supported in Aerospike.\n", "\n", "[Netty](https://netty.io/) is an asynchronous event-driven network application framework for high-performance servers based on Java Non-blocking IO ([NIO](https://en.wikipedia.org/wiki/Non-blocking_I/O_(Java))) package. [Epoll](https://en.wikipedia.org/wiki/Epoll) (event poll)is a Linux specific construct and allows for a process to monitor multiple file descriptors and get notifications when I/O is possible on them. \n", "\n", "Netty allows users to share their existing event loops with AerospikeClient which can improve performance. Netty event loops are also required when using TLS connections. However Netty is an optional external library dependency. \n", "\n", "Direct NIO event loops are lighter weight and slightly faster than Netty defaults when not sharing event loops. Direct NIO does not have an external library dependency.\n", "\n", "You should consider trade-offs in using the types of event loops - refer to the links provided for further details." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Async Framework\n", "Below we walk through the steps in setting up a typical async operation framework." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize event loops\n", "Initialize event loops. Allocate an event loop for each CPU core. \n", "\n", "Examine the code snippets below.\n", "\n", "- Initialize event policy. Select level of parallelism desired; cannot exceed server throughput. \n", "
\n",
    "EventPolicy eventPolicy = EventPolicy();\n",
    "final CommandsPerEventLoop = 50;\n",
    "eventPolicy.maxCommandsInProcess = commandsPerEventLoop;\n",
    "
\n", "- Select delay queue buffer size in front of the event loop. \n", "
\n",
    "maxCommandsInQueue = 50;\n",
    "eventPolicy.maxCommandsInQueue = maxCommandsInQueue;\n",
    "
\n", "- Create event loops object.\n", "
\n",
    "// here we use direct nio and 2 events loops\n",
    "numLoops = 2;\n",
    "EventLoops eventLoops = new NioEventLoops(eventPolicy, numLoops);\n",
    "
\n", "\n", "In the following cell, the function InitializeEventLoops allows initialization of different types of event loops. The function will be called multiple times later in the notebook to experiment with different settings." ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Event loops initialized with num-loops: 2, commands-per-event-loop: 50, delay-queue-size: 50.\n" ] } ], "source": [ "import com.aerospike.client.async.EventPolicy;\n", "import io.netty.channel.nio.NioEventLoopGroup;\n", "import io.netty.channel.epoll.EpollEventLoopGroup;\n", "import com.aerospike.client.async.NettyEventLoops;\n", "import com.aerospike.client.async.EventLoops;\n", "import com.aerospike.client.async.NioEventLoops;\n", "\n", "enum EventLoopType{DIRECT_NIO, NETTY_NIO, NETTY_EPOLL};\n", "\n", "// a function to create event loops with specified parameters\n", "EventLoops InitializeEventLoops(EventLoopType eventLoopType, int numLoops, int commandsPerEventLoop, \n", " int maxCommandsInQueue) {\n", " EventPolicy eventPolicy = new EventPolicy();\n", " eventPolicy.maxCommandsInProcess = commandsPerEventLoop;\n", " eventPolicy.maxCommandsInQueue = maxCommandsInQueue;\n", " EventLoops eventLoops = null;\n", " switch(eventLoopType) {\n", " case DIRECT_NIO:\n", " eventLoops = new NioEventLoops(eventPolicy, numLoops);\n", " break;\n", " case NETTY_NIO:\n", " NioEventLoopGroup nioGroup = new NioEventLoopGroup(numLoops);\n", " eventLoops = new NettyEventLoops(eventPolicy, nioGroup);\n", " break;\n", " case NETTY_EPOLL:\n", " EpollEventLoopGroup epollGroup = new EpollEventLoopGroup(numLoops);\n", " eventLoops = new NettyEventLoops(eventPolicy, epollGroup);\n", " break;\n", " default:\n", " System.out.println(\"Error: Invalid event loop type\");\n", " }\n", " return eventLoops;\n", "}\n", "\n", "// initialize event loops \n", "final int NumLoops = 2;\n", "final int CommandsPerEventLoop = 50;\n", "final int DelayQueueSize = 50;\n", "\n", "EventLoops eventLoops = InitializeEventLoops(EventLoopType.DIRECT_NIO, NumLoops, CommandsPerEventLoop, DelayQueueSize);\n", "\n", "System.out.format(\"Event loops initialized with num-loops: %s, commands-per-event-loop: %s, delay-queue-size: %s.\\n\",\n", " NumLoops, CommandsPerEventLoop, DelayQueueSize);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize Client\n", "Examine the code snippets below.\n", "\n", "- Initialize client policy with event loops.\n", "
\n",
    "ClientPolicy policy = new ClientPolicy();\n",
    "clientPolicy.eventLoops = eventLoops;\n",
    "
\n", " - Set total concurrent connections per node by multiplying concurrency level at event loop (maxCommandsInProcess) by the number of event loops. \n", "
\n",
    "concurrentMax = commandsPerEventLoop * numLoops;\n",
    "
\n", " - This is the max number of commands or requests per node if all requests go to one node. Adjust the default connection pool size of 300 if concurrentMax is larger.\n", "
\n",
    "if (clientPolicy.maxConnsPerNode < concurrentMax) {\n",
    "    clientPolicy.maxConnsPerNode = concurrentMax; \n",
    "}\n",
    "
\n", "- Initialize the client with the client policy and seed hosts in cluster.\n", "
\n",
    "Host[] hosts = Host.parseHosts(\"localhost\", 3000);\n",
    "AerospikeClient client = new AerospikeClient(clientPolicy, hosts);\n",
    "
\n", "\n", "In the following cell, the function InitializeClient allows initialization of the client with specified parameters. " ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Client initialized.\n" ] } ], "source": [ "import com.aerospike.client.policy.ClientPolicy;\n", "import com.aerospike.client.Host;\n", "import com.aerospike.client.AerospikeClient;\n", "\n", "// a function to initialize the client with specified parameters\n", "AerospikeClient InitializeClient(EventLoops eventLoops, int numLoops, int commandsPerEventLoop, Host[] hosts) {\n", " ClientPolicy clientPolicy = new ClientPolicy();\n", " clientPolicy.eventLoops = eventLoops;\n", " int concurrentMax = commandsPerEventLoop * numLoops;\n", " if (clientPolicy.maxConnsPerNode < concurrentMax) {\n", " clientPolicy.maxConnsPerNode = concurrentMax; \n", " }\n", " AerospikeClient client = new AerospikeClient(clientPolicy, hosts);\n", " return client;\n", "}\n", "\n", "// initialize the client \n", "Host[] hosts = Host.parseHosts(\"localhost\", 3000); \n", "\n", "AerospikeClient client = InitializeClient(eventLoops, NumLoops, CommandsPerEventLoop, hosts);\n", "\n", "System.out.print(\"Client initialized.\\n\");" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize event loop throttles and atomic operation count.\n", "The event loop throttles object is initialized with the number of event loops and commands per event loop. It provides two methods \"waitForSlot\" and \"addSlot\" to manage concurrency for an event loop, both take an index parameter that identifies the event loop. \n", "
\n",
    "throttles = new Throttles(numLoops, commandsPerEventLoop);\n",
    "
\n", "\n", "The operation count is used to track the number of finished operations. Because multiple callback threads access and increment it concurrently, it is defined as an AtomicInteger, which has support for atomic operation get/increment operations.\n", "
\n",
    "AtomicInteger asyncOpCount = new AtomicInteger();\n",
    "
\n", "In the following cell, the function InitializeThrottles creates throttles for event loops with specified parameters. " ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Throttles initialized for 2 loops with 50 concurrent operations per loop.\n", "Atomic operation count initialized." ] } ], "source": [ "import com.aerospike.client.async.Throttles;\n", "\n", "// creates event loop throttles with specified parameters\n", "Throttles InitializeThrottles(int numLoops, int commandsPerEventLoop) {\n", " Throttles throttles = new Throttles(numLoops, commandsPerEventLoop);\n", " return throttles;\n", "}\n", "\n", "// initialize event loop throttles\n", "Throttles throttles = InitializeThrottles(NumLoops, CommandsPerEventLoop);\n", "System.out.format(\"Throttles initialized for %s loops with %s concurrent operations per loop.\\n\", \n", " NumLoops, CommandsPerEventLoop);\n", "\n", "// initialize the atomic integer to keep track of async operations count\n", "import java.util.concurrent.atomic.AtomicInteger;\n", "AtomicInteger asyncOpCount = new AtomicInteger();\n", "System.out.format(\"Atomic operation count initialized.\");;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Define Listener and Handlers\n", "Define the listener with success and failure handlers to process results. Below, we have MyWriteListener derived from WriteListener to process insertion of records that: \n", "- implements success and failure handlers\n", "- releases a slot in the event loop on success or failure for another insert to proceed\n", " throttles.addSlot(eventLoopIndex, 1);\n", "- signals completion through monitor on failure or when the write count reaches the expected final count\n", " monitor.notifyComplete();\n", "- prints progress every \"progressFreq\" records" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Write listener defined." ] } ], "source": [ "import com.aerospike.client.Key;\n", "import com.aerospike.client.listener.WriteListener;\n", "import com.aerospike.client.async.Monitor;\n", "import com.aerospike.client.AerospikeException;\n", "\n", "// write listener \n", "// - implements success and failure handlers\n", "// - releases a slot on success or failure for another insert to proceed\n", "// - signals completion through monitor on failure or when the write count reaches the expected final count\n", "// - prints progress every \"progressFreq\" records*/\n", "\n", "class MyWriteListener implements WriteListener {\n", " private final Key key;\n", " private final int eventLoopIndex;\n", " private final int finalCount;\n", " private Monitor monitor; \n", " private final int progressFreq;\n", " \n", " public MyWriteListener(Key key, int eventLoopIndex, int finalCount, Monitor monitor, int progressFreq) {\n", " this.key = key;\n", " this.eventLoopIndex = eventLoopIndex;\n", " this.finalCount = finalCount;\n", " this.monitor = monitor;\n", " this.progressFreq = progressFreq;\n", " }\n", "\n", " // Write success callback.\n", " public void onSuccess(Key key) {\n", " // Write succeeded.\n", " throttles.addSlot(eventLoopIndex, 1);\n", " int currentCount = asyncOpCount.incrementAndGet();\n", " if ( progressFreq > 0 && currentCount % progressFreq == 0) {\n", " System.out.format(\"Inserted %s records.\\n\", currentCount); \n", " }\n", " if (currentCount == finalCount) {\n", " monitor.notifyComplete();\n", " }\n", " }\n", "\n", " // Error callback.\n", " public void onFailure(AerospikeException e) {\n", " System.out.format(\"Put failed: namespace=%s set=%s key=%s exception=%s\\n\", \n", " key.namespace, key.setName, key.userKey, e.getMessage());\n", " monitor.notifyComplete();\n", " }\n", "}\n", "\n", "System.out.print(\"Write listener defined.\"); " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Submit Async Requests Using Throttling\n", "While submitting async requests it is important to keep below the planned concurrent capacity using throttling.\n", "\n", "The function InsertRecords below inserts the specified number of records asynchronously with id-\\ as the user-key and two integer fields bin1 and bin2. It keeps track of and returns the elapsed time.\n", "\n", "Throttling is achieved by waiting for an available slot in the event loop.\n", "
\n",
    "if (throttles.waitForSlot(eventLoopIndex, 1)) {\n",
    "    // submit async request\n",
    "}\n",
    "
\n", "\n", "After submitting all requests, the main thread must wait for outstanding requests to complete before closing.\n", "
\n",
    "   monitor.waitTillComplete();\n",
    "
" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inserted 25000 records.\n", "Inserted 50000 records.\n", "Inserted 75000 records.\n", "Inserted 100000 records.\n", "Inserted 100000 records with 2 event-loops and 50 commands-per-loop in 4212 milliseconds.\n" ] } ], "source": [ "import java.util.concurrent.TimeUnit;\n", "import com.aerospike.client.Bin;\n", "import com.aerospike.client.policy.WritePolicy;\n", "import com.aerospike.client.async.EventLoop;\n", "\n", "long InsertRecords(int numRecords, EventLoops eventLoops, Throttles throttles, int progressFreq) {\n", " long startTime = System.nanoTime();\n", " Monitor monitor = new Monitor();\n", " asyncOpCount.set(0);\n", " WritePolicy policy = new WritePolicy();\n", "\n", " for (int i = 0; i < numRecords; i++) {\n", " Key key = new Key(Namespace, Set, \"id-\"+i);\n", " Bin bin1 = new Bin(new String(\"bin1\"), i);\n", " Bin bin2 = new Bin(new String(\"bin2\"), numRecords*10+i);\n", " EventLoop eventLoop = eventLoops.next();\n", " int eventLoopIndex = eventLoop.getIndex();\n", "\n", " if (throttles.waitForSlot(eventLoopIndex, 1)) {\n", " try {\n", " client.put(eventLoop, new MyWriteListener(key, eventLoopIndex, numRecords, monitor, progressFreq), \n", " policy, key, bin1, bin2);\n", " }\n", " catch (Exception e) {\n", " throttles.addSlot(eventLoopIndex, 1);\n", " }\n", " }\n", " }\n", " monitor.waitTillComplete(); \n", " long endTime = System.nanoTime();\n", " return (endTime - startTime);\n", "}\n", "\n", "final int NumRecords = 100000;\n", "\n", "long elapsedTime = InsertRecords(NumRecords, eventLoops, throttles, NumRecords/4);\n", "\n", "System.out.format(\"Inserted %s records with %s event-loops and %s commands-per-loop in %s milliseconds.\\n\", \n", " NumRecords, NumLoops, CommandsPerEventLoop, elapsedTime/1000000);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Closing\n", "Both AerospikeClient and EventLoops should be closed before program shutdown. The latest client waits for pending async commands to finish before performing the actual close, so there is no need to externally track pending async commands. Earlier versions provide a waitToComplete() call on Monitor object to ensure async operations are completed. The Cleanup function implemented above truncates the database and closes client and event-loops." ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Removed data and closed client and event loops.\n" ] } ], "source": [ "// truncates database and closes client and event-loops\n", "Cleanup();\n", "System.out.println(\"Removed data and closed client and event loops.\");" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Nested and Inline Async Operations\n", "It is possible to nest a series of async calls, one in the processing logic of another. Some simple examples of such cascaded calls are:\n", "- Retry the same operation in the failure handler\n", "- Issue an async read to validate an async write operation\n", "- Issue an async write to update a record retrieved from an async read operation.\n", "\n", "The following code illustrates a simplistic example of how each record retrieved from an async filtered scan is updated asynchronously by incrementing the value of bin2. Note the inline implementation of WriteListener. The scan filter selects records between bin1 values of 1 and 1000. Throttling and progress report are also present as described above." ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "import com.aerospike.client.policy.ScanPolicy;\n", "import com.aerospike.client.listener.RecordSequenceListener;\n", "import com.aerospike.client.Record;\n", "import com.aerospike.client.exp.Exp;\n", "\n", "// Scan callback\n", "class ScanRecordSequenceListener implements RecordSequenceListener {\n", " private EventLoops eventLoops;\n", " private Throttles throttles;\n", " private Monitor scanMonitor;\n", " private AtomicInteger writeCount = new AtomicInteger();\n", " private int scanCount = 0; \n", " private final int progressFreq;\n", " \n", " public ScanRecordSequenceListener(EventLoops eventLoops, Throttles throttles, Monitor scanMonitor, \n", " int progressFreq) {\n", " this.eventLoops = eventLoops;\n", " this.throttles = throttles;\n", " this.scanMonitor = scanMonitor;\n", " this.progressFreq = progressFreq;\n", " }\n", " \n", " public void onRecord(Key key, Record record) throws AerospikeException {\n", " ++scanCount;\n", " if ( progressFreq > 0 && scanCount % progressFreq == 0) {\n", " System.out.format(\"Scan returned %s records.\\n\", scanCount); \n", " }\n", " // submit async update operation with throttle\n", " EventLoop eventLoop = eventLoops.next();\n", " int eventLoopIndex = eventLoop.getIndex();\n", " if (throttles.waitForSlot(eventLoopIndex, 1)) { // throttle by waiting for an available slot\n", " try {\n", " WritePolicy policy = new WritePolicy();\n", " Bin bin2 = new Bin(new String(\"bin2\"), 1);\n", " \n", " client.add(eventLoop, new WriteListener() { // inline write listener\n", " \n", " public void onSuccess(final Key key) {\n", " // Write succeeded.\n", " throttles.addSlot(eventLoopIndex, 1);\n", " int currentCount = writeCount.incrementAndGet();\n", " if ( progressFreq > 0 && currentCount % progressFreq == 0) {\n", " System.out.format(\"Processed %s records.\\n\", currentCount); \n", " }\n", " }\n", "\n", " public void onFailure(AerospikeException e) {\n", " System.out.format(\"Put failed: namespace=%s set=%s key=%s exception=%s\\n\", \n", " key.namespace, key.setName, key.userKey, e.getMessage());\n", " throttles.addSlot(eventLoopIndex, 1);\n", " int currentCount = writeCount.incrementAndGet();\n", " if ( progressFreq > 0 && currentCount % progressFreq == 0) {\n", " System.out.format(\"Processed %s records.\\n\", currentCount); \n", " }\n", " }\n", " }, \n", " policy, key, bin2);\n", " }\n", " catch (Exception e) {\n", " System.out.format(\"Error: exception in write listener - %s\", e.getMessage());\n", " }\n", " }\n", " }\n", "\n", " public void onSuccess() {\n", " if (scanCount != writeCount.get()) { // give the last write some time to finish\n", " try {\n", " Thread.sleep(100);\n", " } \n", " catch(InterruptedException e) {\n", " System.out.format(\"Error: exception - %s\", e);\n", " }\n", " }\n", " scanMonitor.notifyComplete();\n", " }\n", "\n", " public void onFailure(AerospikeException e) {\n", " System.out.format(\"Error: scan failed with exception - %s\", e);\n", " scanMonitor.notifyComplete();\n", " }\n", "}" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inserted 100000 records.\n", "Scan returned 100 records.\n", "Processed 100 records.\n", "Scan returned 200 records.\n", "Processed 200 records.\n", "Scan returned 300 records.\n", "Processed 300 records.\n", "Scan returned 400 records.\n", "Processed 400 records.\n", "Scan returned 500 records.\n", "Processed 500 records.\n", "Scan returned 600 records.\n", "Processed 600 records.\n", "Scan returned 700 records.\n", "Processed 700 records.\n", "Scan returned 800 records.\n", "Processed 800 records.\n", "Scan returned 900 records.\n", "Processed 900 records.\n", "Scan returned 1000 records.\n", "Processed 1000 records.\n", "Done: nested async scan and update" ] } ], "source": [ "// cleanup prior state\n", "Cleanup();\n", "\n", "// initialize data, event loops and client\n", "int numRecords = 100000;\n", "int numLoops = 2;\n", "int commandsPerLoop = 25;\n", "int delayQueueSize = 0;\n", "\n", "eventLoops = InitializeEventLoops(EventLoopType.DIRECT_NIO, numLoops, commandsPerLoop, delayQueueSize);\n", "client = InitializeClient(eventLoops, numLoops, commandsPerLoop, hosts);\n", "throttles = InitializeThrottles(numLoops, commandsPerLoop);\n", "\n", "InsertRecords(numRecords, eventLoops, throttles, 0);\n", "System.out.format(\"Inserted %s records.\\n\", numRecords);\n", "\n", "EventLoop eventLoop = eventLoops.next();\n", "Monitor scanMonitor = new Monitor();\n", "int progressFreq = 100;\n", "\n", "// issue async scan that in turn issues async update on each returned record\n", "ScanPolicy policy = new ScanPolicy();\n", "policy.filterExp = Exp.build(\n", " Exp.and(\n", " Exp.le(Exp.intBin(\"bin1\"), Exp.val(1000)),\n", " Exp.ge(Exp.intBin(\"bin1\"), Exp.val(1))));\n", " \n", "client.scanAll(eventLoop, new ScanRecordSequenceListener(eventLoops, throttles, scanMonitor, progressFreq), \n", " policy, Namespace, Set);\n", "scanMonitor.waitTillComplete();\n", "\n", "System.out.format(\"Done: nested async scan and update\");;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Misc Examples" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Delay Queue Full Error\n", "If the delay queue fills up, a request will not be accepted and the client will return “delay queue full” error. Below we simulate this condition by having 25 slots and a delay queue of 20 in 2 event loops each (can handle total 90 outstanding requests) and issuing a hundred concurrent requests. The throttle is effectively turned off by a large setting for the number of requests to go through." ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Put failed: namespace=test set=async-ops key=id-90 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-92 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-91 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-93 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-95 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-97 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-99 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-94 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-96 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "Put failed: namespace=test set=async-ops key=id-98 exception=Error -9,0,30000,0,0: Async delay queue is full\n", "16 ops/ms with event-loops: 2 and commands-per-loop: 25.\n" ] } ], "source": [ "// clean up the current state\n", "Cleanup();\n", "\n", "// initialize data, event loops and client\n", "int numRecords = 100;\n", "int numLoops = 2;\n", "int commandsPerLoop = 25;\n", "int delayQueueSize = 20;\n", "int noThrottle = 10000; //effectively no throttle\n", "\n", "eventLoops = InitializeEventLoops(EventLoopType.DIRECT_NIO, numLoops, commandsPerLoop, delayQueueSize);\n", "client = InitializeClient(eventLoops, numLoops, commandsPerLoop, hosts);\n", "throttles = InitializeThrottles(numLoops, noThrottle);\n", "\n", "// attempt to insert records above the available slots and delay queue capacity\n", "long elapsedTime = InsertRecords(numRecords, eventLoops, throttles, 0);\n", "\n", "System.out.format(\"%s ops/ms with event-loops: %s and commands-per-loop: %s.\\n\", \n", " numRecords/(elapsedTime/1000000), numLoops, commandsPerLoop);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Comparing Different Settings\n", "The code below allows comparison of insert throughput with different parameters: event loops type, number of event loops, and concurrency level in each loop. It doesn't produce meaningful results in the default notebook container setting where the client and server are running in the same container. A meaningful comparison can be drawn by pointing to the desired server cluster and also adjusting the client environment." ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "27 ops/ms with 2 DIRECT_NIO event-loops and 50 commands-per-loop.\n", "23 ops/ms with 2 DIRECT_NIO event-loops and 100 commands-per-loop.\n", "29 ops/ms with 2 DIRECT_NIO event-loops and 200 commands-per-loop.\n", "34 ops/ms with 4 DIRECT_NIO event-loops and 50 commands-per-loop.\n", "35 ops/ms with 4 DIRECT_NIO event-loops and 100 commands-per-loop.\n", "33 ops/ms with 4 DIRECT_NIO event-loops and 200 commands-per-loop.\n", "23 ops/ms with 8 DIRECT_NIO event-loops and 50 commands-per-loop.\n", "22 ops/ms with 8 DIRECT_NIO event-loops and 100 commands-per-loop.\n", "19 ops/ms with 8 DIRECT_NIO event-loops and 200 commands-per-loop.\n", "21 ops/ms with 2 NETTY_NIO event-loops and 50 commands-per-loop.\n", "25 ops/ms with 2 NETTY_NIO event-loops and 100 commands-per-loop.\n", "27 ops/ms with 2 NETTY_NIO event-loops and 200 commands-per-loop.\n", "32 ops/ms with 4 NETTY_NIO event-loops and 50 commands-per-loop.\n", "31 ops/ms with 4 NETTY_NIO event-loops and 100 commands-per-loop.\n", "33 ops/ms with 4 NETTY_NIO event-loops and 200 commands-per-loop.\n", "22 ops/ms with 8 NETTY_NIO event-loops and 50 commands-per-loop.\n", "22 ops/ms with 8 NETTY_NIO event-loops and 100 commands-per-loop.\n", "20 ops/ms with 8 NETTY_NIO event-loops and 200 commands-per-loop.\n", "20 ops/ms with 2 NETTY_EPOLL event-loops and 50 commands-per-loop.\n", "23 ops/ms with 2 NETTY_EPOLL event-loops and 100 commands-per-loop.\n", "24 ops/ms with 2 NETTY_EPOLL event-loops and 200 commands-per-loop.\n", "25 ops/ms with 4 NETTY_EPOLL event-loops and 50 commands-per-loop.\n", "25 ops/ms with 4 NETTY_EPOLL event-loops and 100 commands-per-loop.\n", "26 ops/ms with 4 NETTY_EPOLL event-loops and 200 commands-per-loop.\n", "23 ops/ms with 8 NETTY_EPOLL event-loops and 50 commands-per-loop.\n", "23 ops/ms with 8 NETTY_EPOLL event-loops and 100 commands-per-loop.\n", "21 ops/ms with 8 NETTY_EPOLL event-loops and 200 commands-per-loop.\n", "Done.\n" ] } ], "source": [ "// Throughput with parameterized async insertion\n", "\n", "int numRecords = 100000;\n", "EventLoopType[] eventLoopOptions = {EventLoopType.DIRECT_NIO, EventLoopType.NETTY_NIO, EventLoopType.NETTY_EPOLL};\n", "int[] numLoopsOptions = {2, 4, 8};\n", "int[] commandsPerLoopOptions = {50, 100, 200};\n", "\n", "for (EventLoopType eventLoopType: eventLoopOptions) {\n", " for (int numLoops: numLoopsOptions) {\n", " for (int commandsPerLoop: commandsPerLoopOptions) {\n", " Cleanup();\n", " eventLoops = InitializeEventLoops(eventLoopType, numLoops, commandsPerLoop, 0);\n", " client = InitializeClient(eventLoops, numLoops, commandsPerLoop, hosts);\n", " throttles = InitializeThrottles(numLoops, commandsPerLoop);\n", "\n", " long elapsedTime = InsertRecords(numRecords, eventLoops, throttles, 0);\n", " System.out.format(\"%s ops/ms with %s %s event-loops and %s commands-per-loop.\\n\", \n", " numRecords/(elapsedTime/1000000), numLoops, eventLoopType, commandsPerLoop);\n", " }\n", " }\n", "}\n", "System.out.println(\"Done.\");;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Takeaways and Conclusion\n", "The tutorial described the architecture of and key concepts in asynchronous operations in Aerospike client. It presented the programming framework in which async requests can be submitted and handled. It illustrated with code how event loops, throttling, inline async calls are implemented. The trade-offs that a developer needs to make for which execution modes to employ - synchronous, asynchronous, or background - involve multiple factors including the nature of operations, client and server setup, throughput needs, and programming complexity." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Clean up\n", "Remove tutorial data and close connection." ] }, { "cell_type": "code", "execution_count": 14, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:49:19.972650Z", "start_time": "2020-12-29T20:49:19.967344Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Removed tutorial data and closed server connection.\n" ] } ], "source": [ "Cleanup();\n", "System.out.println(\"Removed tutorial data and closed server connection.\");" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Further Exploration and Resources\n", "Here are some links for further exploration\n", "\n", "Resources\n", "- Related notebooks\n", " - [Implementing SQL Operations: SELECT](sql_select.ipynb), \n", " - [Implementing SQL Operations: Aggregates - Part 1](sql_aggregates_1.ipynb) and [Part 2](sql_aggregates_2.ipynb).\n", " - [Implementing SQL Operations: CREATE, UPDATE, DELETE](sql_updates.ipynb)\n", " - [Working with Lists](java-working_with_lists.ipynb)\n", " - [Working with Maps](java-working_with_maps.ipynb)\n", "- Aerospike Developer Hub\n", " - [Java Developers Resources](https://developer.aerospike.com/java-developers)\n", "- Github repos\n", " - [Java code examples](https://github.com/aerospike/aerospike-client-java/tree/master/examples/src/com/aerospike/examples)\n", " - [Reactive programming examples for the Java client](https://github.com/aerospike/aerospike-client-java-reactive)\n", "- Documentation\n", " - [Java Client](https://www.aerospike.com/docs/client/java/index.html)\n", " - [Java API Reference](https://www.aerospike.com/apidocs/java/)\n", " - [Aerospike Documentation](https://docs.aerospike.com/docs/)\n", "- Blog\n", " - [Simple Web Application Using Java, Spring Boot, Aerospike and Docker](https://medium.com/aerospike-developer-blog/simple-web-application-using-java-spring-boot-aerospike-database-and-docker-ad13795e0089)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next steps\n", "\n", "Visit [Aerospike notebooks repo](https://github.com/aerospike-examples/interactive-notebooks) to run additional Aerospike notebooks. To run a different notebook, download the notebook from the repo to your local machine, and then click on File->Open in the notebook menu, and select Upload." ] } ], "metadata": { "kernelspec": { "display_name": "Java", "language": "java", "name": "java" }, "language_info": { "codemirror_mode": "java", "file_extension": ".jshell", "mimetype": "text/x-java-source", "name": "Java", "pygments_lexer": "java", "version": "11.0.8+10-LTS" }, "toc": { "base_numbering": 1, "nav_menu": {}, "number_sections": true, "sideBar": true, "skip_h1_title": false, "title_cell": "Table of Contents", "title_sidebar": "Contents", "toc_cell": true, "toc_position": {}, "toc_section_display": true, "toc_window_display": false } }, "nbformat": 4, "nbformat_minor": 4 }