Implements SDKCtlSocket that implements communication protocol wih SdkController
In addition, this CL contains some minor tweaks to async-socket, and
async-socket-connector that improve tracebility.
Change-Id: Ib1309b19dcd02e96379155fea7015019d93160e7
diff --git a/android/sdk-controller-socket.c b/android/sdk-controller-socket.c
new file mode 100644
index 0000000..a300169
--- /dev/null
+++ b/android/sdk-controller-socket.c
@@ -0,0 +1,1691 @@
+/*
+ * Copyright (C) 2012 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Encapsulates exchange protocol between the emulator, and an Android device
+ * that is connected to the host via USB. The communication is established over
+ * a TCP port forwarding, enabled by ADB.
+ */
+
+#include "qemu-common.h"
+#include "android/async-utils.h"
+#include "android/utils/debug.h"
+#include "android/async-socket-connector.h"
+#include "android/async-socket.h"
+#include "android/sdk-controller-socket.h"
+#include "utils/panic.h"
+#include "iolooper.h"
+
+#define E(...) derror(__VA_ARGS__)
+#define W(...) dwarning(__VA_ARGS__)
+#define D(...) VERBOSE_PRINT(sdkctlsocket,__VA_ARGS__)
+#define D_ACTIVE VERBOSE_CHECK(sdkctlsocket)
+
+#define TRACE_ON 1
+
+#if TRACE_ON
+#define T(...) VERBOSE_PRINT(sdkctlsocket,__VA_ARGS__)
+#else
+#define T(...)
+#endif
+
+/* Recycling memory descriptor. */
+typedef struct SDKCtlRecycled SDKCtlRecycled;
+struct SDKCtlRecycled {
+ union {
+ /* Next recycled descriptor (while listed in recycler). */
+ SDKCtlRecycled* next;
+ /* Allocated memory size (while outside of the recycler). */
+ uint32_t size;
+ };
+};
+
+/********************************************************************************
+ * SDKCtlPacket declarations
+ *******************************************************************************/
+
+/*
+ * Types of the packets of data sent via SDK controller socket.
+ */
+
+/* The packet is a message. */
+#define SDKCTL_PACKET_MESSAGE 1
+/* The packet is a query. */
+#define SDKCTL_PACKET_QUERY 2
+/* The packet is a response to a query. */
+#define SDKCTL_PACKET_QUERY_RESPONSE 3
+
+/* Data packet descriptor.
+ *
+ * All packets, sent and received via SDK controller socket begin with this
+ * header, with packet data immediately following this header.
+ */
+typedef struct SDKCtlPacketHeader {
+ /* Total size of the data to transfer with this packet, including this
+ * header. The transferring data should immediatelly follow this header. */
+ int size;
+ /* Encodes packet type. See SDKCTL_PACKET_XXX for the list of packet types
+ * used by SDK controller. */
+ int type;
+} SDKCtlPacketHeader;
+
+/* Packet descriptor, allocated by this API for data packets to be sent to SDK
+ * controller service on the device.
+ *
+ * When packet descriptors are allocated by this API, they are allocated large
+ * enough to contain this header, and packet data to send to the service,
+ * immediately following this descriptor.
+ */
+struct SDKCtlPacket {
+ /* Supports recycling. Don't put anything in front: recycler expects this
+ * to be the first field in recyclable descriptor. */
+ SDKCtlRecycled recycling;
+
+ /* Next packet in the list of packets to send. */
+ SDKCtlPacket* next;
+ /* SDK controller socket that transmits this packet. */
+ SDKCtlSocket* sdkctl;
+ /* Number of outstanding references to the packet. */
+ int ref_count;
+
+ /* Common packet header. Packet data immediately follows this header, so it
+ * must be last field in SDKCtlPacket descriptor. */
+ SDKCtlPacketHeader header;
+};
+
+/********************************************************************************
+ * SDKCtlQuery declarations
+ *******************************************************************************/
+
+/*
+ * Types of queries sent via SDK controller socket.
+ */
+
+/* Handshake query.
+ * This query is sent to SDK controller service as part of the connection
+ * protocol implementation.
+ */
+#define SDKCTL_QUERY_HANDSHAKE -1
+
+/* Query packet descriptor.
+ *
+ * All queries, sent and received via SDK controller socket begin with this
+ * header, with query data immediately following this header.
+ */
+typedef struct SDKCtlQueryHeader {
+ /* Data packet header for this query. */
+ SDKCtlPacketHeader packet;
+ /* A unique query identifier. This ID is used to track the query in the
+ * asynchronous environment in whcih SDK controller socket operates. */
+ int query_id;
+ /* Query type. See SDKCTL_QUERY_XXX for the list of query types used by SDK
+ * controller. */
+ int query_type;
+} SDKCtlQueryHeader;
+
+/* Query descriptor, allocated by this API for queries to be sent to SDK
+ * controller service on the device.
+ *
+ * When query descriptors are allocated by this API, they are allocated large
+ * enough to contain this header, and query data to send to the service,
+ * immediately following this descriptor.
+ */
+struct SDKCtlQuery {
+ /* Supports recycling. Don't put anything in front: recycler expects this
+ * to be the first field in recyclable descriptor. */
+ SDKCtlRecycled recycling;
+
+ /* Next query in the list of active, or recycled queries. */
+ SDKCtlQuery* next;
+ /* A timer to run time out on this query after it has been sent. */
+ LoopTimer timer[1];
+ /* Absolute time for this query's deadline. This is the value that query's
+ * timer is set for after query has been transmitted to the service. */
+ Duration deadline;
+ /* SDK controller socket that owns the query. */
+ SDKCtlSocket* sdkctl;
+ /* A callback to invoke on query state changes. */
+ on_sdkctl_query_cb query_cb;
+ /* An opaque pointer associated with this query. */
+ void* query_opaque;
+ /* Points to an address of a buffer where to save query response. */
+ void** response_buffer;
+ /* Points to a variable containing size of the response buffer (on the way in),
+ * or actual query response size (when query is completed). */
+ uint32_t* response_size;
+ /* Internal response buffer, allocated if query creator didn't provide its
+ * own. This field is valid only if response_buffer field is NULL, or is
+ * pointing to this field. */
+ void* internal_resp_buffer;
+ /* Internal response buffer size This field is valid only if response_size
+ * field is NULL, or is pointing to this field. */
+ uint32_t internal_resp_size;
+ /* Number of outstanding references to the query. */
+ int ref_count;
+
+ /* Common packet header. Query data immediately follows this header, so it
+ * must be last field in SDKCtlQuery descriptor. */
+ SDKCtlQueryHeader header;
+};
+
+/* Query reply descriptor.
+ *
+ * All replies to a query, sent and received via SDK controller socket begin with
+ * this header, with query reply data immediately following this header.
+ */
+typedef struct SDKCtlQueryReplyHeader {
+ /* Data packet header for this reply. */
+ SDKCtlPacketHeader packet;
+
+ /* An identifier for the query that is addressed with this reply. */
+ int query_id;
+} SDKCtlQueryReplyHeader;
+
+/********************************************************************************
+ * SDK Control Socket declarations
+ *******************************************************************************/
+
+/* Enumerates SDKCtlSocket states. */
+typedef enum SDKCtlSocketState {
+ /* Socket is disconnected from SDK controller. */
+ SDKCTL_SOCKET_DISCONNECTED,
+ /* Connection to SDK controller is in progress. */
+ SDKCTL_SOCKET_CONNECTING,
+ /* Socket is connected to an SDK controller service. */
+ SDKCTL_SOCKET_CONNECTED
+} SDKCtlSocketState;
+
+/* Enumerates SDKCtlSocket I/O dispatcher states. */
+typedef enum SDKCtlIODispatcherState {
+ /* I/O dispatcher expects a packet header. */
+ SDKCTL_IODISP_EXPECT_HEADER,
+ /* I/O dispatcher expects packet data. */
+ SDKCTL_IODISP_EXPECT_DATA,
+ /* I/O dispatcher expects query response header. */
+ SDKCTL_IODISP_EXPECT_QUERY_REPLY_HEADER,
+ /* I/O dispatcher expects query response data. */
+ SDKCTL_IODISP_EXPECT_QUERY_REPLY_DATA,
+} SDKCtlIODispatcherState;
+
+/* SDKCtlSocket I/O dispatcher descriptor. */
+typedef struct SDKCtlIODispatcher {
+ /* SDKCtlSocket instance for this dispatcher. */
+ SDKCtlSocket* sdkctl;
+ /* Dispatcher state. */
+ SDKCtlIODispatcherState state;
+ /* Unites all types of headers used in SDK controller data exchange. */
+ union {
+ /* Common packet header. */
+ SDKCtlPacketHeader header;
+ /* Header for a query packet. */
+ SDKCtlQueryHeader query_header;
+ /* Header for a query response packet. */
+ SDKCtlQueryReplyHeader query_reply_header;
+ };
+ /* Descriptor of a packet packet received from SDK controller. */
+ SDKCtlPacket* packet;
+ /* A query for which a reply is currently being received. */
+ SDKCtlQuery* current_query;
+} SDKCtlIODispatcher;
+
+/* SDK controller socket descriptor. */
+struct SDKCtlSocket {
+ /* SDK controller socket state */
+ SDKCtlSocketState state;
+ /* I/O dispatcher for the socket. */
+ SDKCtlIODispatcher io_dispatcher;
+ /* Asynchronous socket connected to SDK Controller on the device. */
+ AsyncSocket* as;
+ /* Client callback that monitors this socket connection. */
+ on_sdkctl_connection_cb on_connection;
+ /* A callback to invoke when handshake message is received from the
+ * SDK controller. */
+ on_sdkctl_handshake_cb on_handshake;
+ /* A callback to invoke when a message is received from the SDK controller. */
+ on_sdkctl_message_cb on_message;
+ /* An opaque pointer associated with this socket. */
+ void* opaque;
+ /* Name of an SDK controller service this socket is connected to. */
+ char* service_name;
+ /* I/O looper for timers. */
+ Looper* looper;
+ /* Head of the active query list. */
+ SDKCtlQuery* query_head;
+ /* Tail of the active query list. */
+ SDKCtlQuery* query_tail;
+ /* Query ID generator that gets incremented for each new query. */
+ int next_query_id;
+ /* Timeout before trying to reconnect after disconnection. */
+ int reconnect_to;
+ /* Number of outstanding references to this descriptor. */
+ int ref_count;
+ /* Head of the recycled memory */
+ SDKCtlRecycled* recycler;
+ /* Recyclable block size. */
+ uint32_t recycler_block_size;
+ /* Maximum number of blocks to recycle. */
+ int recycler_max;
+ /* Number of blocs in the recycler. */
+ int recycler_count;
+};
+
+/********************************************************************************
+ * SDKCtlSocket recycling management
+ *******************************************************************************/
+
+/* Gets a recycled block for a given SDKCtlSocket, or allocates new memory
+ * block. */
+static void*
+_sdkctl_socket_alloc_recycler(SDKCtlSocket* sdkctl, uint32_t size)
+{
+ SDKCtlRecycled* block = NULL;
+
+ if (sdkctl->recycler != NULL && size <= sdkctl->recycler_block_size) {
+ /* There are blocks in the recycler, and requested size fits. */
+ block = sdkctl->recycler;
+ sdkctl->recycler = block->next;
+ block->size = sdkctl->recycler_block_size;
+ sdkctl->recycler_count--;
+ } else if (size <= sdkctl->recycler_block_size) {
+ /* There are no blocks in the recycler, but requested size fits. */
+ block = malloc(sdkctl->recycler_block_size);
+ if (block == NULL) {
+ APANIC("SDKCtl %s: Unable to allocate %d bytes block",
+ sdkctl->service_name, sdkctl->recycler_block_size);
+ }
+ block->size = sdkctl->recycler_block_size;
+ } else {
+ /* Requested size doesn't fit the recycler. */
+ block = malloc(size);
+ if (block == NULL) {
+ APANIC("SDKCtl %s: Unable to allocate %d bytes block",
+ sdkctl->service_name, size);
+ }
+ block->size = size;
+ }
+
+ return block;
+}
+
+/* Recycles, or frees a block of memory for a given SDKCtlSocket. */
+static void
+_sdkctl_socket_free_recycler(SDKCtlSocket* sdkctl, void* mem)
+{
+ SDKCtlRecycled* block = (SDKCtlRecycled*)mem;
+
+ if (sdkctl->recycler_count == sdkctl->recycler_max ||
+ block->size != sdkctl->recycler_block_size) {
+ /* Recycler is full, or block cannot be recycled. */
+ free(mem);
+ return;
+ }
+
+ block->next = sdkctl->recycler;
+ sdkctl->recycler = block;
+ sdkctl->recycler_count++;
+}
+
+/* Empties the recycler for a given SDKCtlSocket. */
+static void
+_sdkctl_socket_empty_recycler(SDKCtlSocket* sdkctl)
+{
+ SDKCtlRecycled* block = sdkctl->recycler;
+ while (block != NULL) {
+ void* to_free = block;
+ block = block->next;
+ free(to_free);
+ }
+ sdkctl->recycler = NULL;
+ sdkctl->recycler_count = 0;
+}
+
+/********************************************************************************
+ * SDKCtlSocket query list management
+ *******************************************************************************/
+
+/* Adds a query to the list of active queries.
+ * Param:
+ * sdkctl - SDKCtlSocket instance for the query.
+ * query - Query to add to the list.
+ */
+static void
+_sdkctl_socket_add_query(SDKCtlQuery* query)
+{
+ SDKCtlSocket* const sdkctl = query->sdkctl;
+ if (sdkctl->query_head == NULL) {
+ sdkctl->query_head = sdkctl->query_tail = query;
+ } else {
+ sdkctl->query_tail->next = query;
+ sdkctl->query_tail = query;
+ }
+
+ /* Keep the query referenced while it's in the list. */
+ sdkctl_query_reference(query);
+}
+
+/* Removes a query from the list of active queries.
+ * Param:
+ * query - Query to remove from the list of active queries.
+ * Return:
+ * Boolean: 1 if query has been removed, or 0 if query has not been found in the
+ * list of active queries.
+ */
+static int
+_sdkctl_socket_remove_query(SDKCtlQuery* query)
+{
+ SDKCtlSocket* const sdkctl = query->sdkctl;
+ SDKCtlQuery* prev = NULL;
+ SDKCtlQuery* head = sdkctl->query_head;
+
+ /* Quick check: the query could be currently handled by dispatcher. */
+ if (sdkctl->io_dispatcher.current_query == query) {
+ /* Release the query from dispatcher. */
+ sdkctl_query_release(query);
+ sdkctl->io_dispatcher.current_query = NULL;
+ return 1;
+ }
+
+ /* Remove query from the list. */
+ while (head != NULL && query != head) {
+ prev = head;
+ head = head->next;
+ }
+ if (head != NULL) {
+ if (prev == NULL) {
+ /* Query is at the head of the list. */
+ assert(query == sdkctl->query_head);
+ sdkctl->query_head = query->next;
+ } else {
+ /* Query is in the middle / at the end of the list. */
+ assert(query != sdkctl->query_head);
+ prev->next = query->next;
+ }
+ if (sdkctl->query_tail == query) {
+ /* Query is at the tail of the list. */
+ assert(query->next == NULL);
+ sdkctl->query_tail = prev;
+ }
+ query->next = NULL;
+
+ /* Release query that is now removed from the list. Note that query
+ * passed to this routine should hold an extra reference, owned by the
+ * caller. */
+ sdkctl_query_release(query);
+ return 1;
+ } else {
+ D("%s: Query %p is not found in the list.", sdkctl->service_name, query);
+ return 0;
+ }
+}
+
+/* Removes a query (based on query ID) from the list of active queries.
+ * Param:
+ * sdkctl - SDKCtlSocket instance that owns the query.
+ * query_id - Identifies the query to remove.
+ * Return:
+ * A query removed from the list of active queries, or NULL if query with the
+ * given ID has not been found in the list.
+ */
+static SDKCtlQuery*
+_sdkctl_socket_remove_query_id(SDKCtlSocket* sdkctl, int query_id)
+{
+ SDKCtlQuery* prev = NULL;
+ SDKCtlQuery* head = sdkctl->query_head;
+
+ /* Quick check: the query could be currently handled by dispatcher. */
+ if (sdkctl->io_dispatcher.current_query != NULL &&
+ sdkctl->io_dispatcher.current_query->header.query_id == query_id) {
+ /* Release the query from dispatcher. */
+ SDKCtlQuery* const query = sdkctl->io_dispatcher.current_query;
+ sdkctl->io_dispatcher.current_query = NULL;
+ return query;
+ }
+
+ /* Remove query from the list. */
+ while (head != NULL && head->header.query_id != query_id) {
+ prev = head;
+ head = head->next;
+ }
+ if (head != NULL) {
+ /* Query is found in the list. */
+ SDKCtlQuery* const query = head;
+ if (prev == NULL) {
+ /* Query is at the head of the list. */
+ assert(query == sdkctl->query_head);
+ sdkctl->query_head = query->next;
+ } else {
+ /* Query is in the middle, or at the end of the list. */
+ assert(query != sdkctl->query_head);
+ prev->next = query->next;
+ }
+ if (sdkctl->query_tail == query) {
+ /* Query is at the tail of the list. */
+ assert(query->next == NULL);
+ sdkctl->query_tail = prev;
+ }
+ query->next = NULL;
+ return query;
+ } else {
+ D("%s: Query ID %d is not found in the list.",
+ sdkctl->service_name, query_id);
+ return NULL;
+ }
+}
+
+/* Pulls the first query from the list of active queries.
+ * Param:
+ * sdkctl - SDKCtlSocket instance that owns the query.
+ * Return:
+ * A query removed pulled from the list of active queries, or NULL if query
+ * list is empty.
+ */
+static SDKCtlQuery*
+_sdkctl_socket_pull_first_query(SDKCtlSocket* sdkctl)
+{
+ SDKCtlQuery* const query = sdkctl->query_head;
+
+ if (query != NULL) {
+ sdkctl->query_head = query->next;
+ if (sdkctl->query_head == NULL) {
+ sdkctl->query_tail = NULL;
+ }
+ }
+ return query;
+}
+
+/* Generates new query ID for the given SDKCtl. */
+static int
+_sdkctl_socket_next_query_id(SDKCtlSocket* sdkctl)
+{
+ return ++sdkctl->next_query_id;
+}
+
+/********************************************************************************
+ * SDKCtlPacket implementation
+ *******************************************************************************/
+
+/* Alocates a packet. */
+static SDKCtlPacket*
+_sdkctl_packet_new(SDKCtlSocket* sdkctl, int size, int type)
+{
+ const uint32_t total_size = sizeof(SDKCtlPacket) + size;
+ SDKCtlPacket* const packet = _sdkctl_socket_alloc_recycler(sdkctl, total_size);
+
+ packet->sdkctl = sdkctl;
+ packet->ref_count = 1;
+ packet->header.size = size;
+ packet->header.type = type;
+
+ /* Refence SDKCTlSocket that owns this packet. */
+ sdkctl_socket_reference(sdkctl);
+
+ return packet;
+}
+
+/* Frees a packet. */
+static void
+_sdkctl_packet_free(SDKCtlPacket* packet)
+{
+ SDKCtlSocket* const sdkctl = packet->sdkctl;
+
+ /* Free allocated resources. */
+ _sdkctl_socket_free_recycler(packet->sdkctl, packet);
+
+ /* Release SDKCTlSocket that owned this packet. */
+ sdkctl_socket_release(sdkctl);
+}
+
+int
+sdkctl_packet_reference(SDKCtlPacket* packet)
+{
+ assert(packet->ref_count > 0);
+ packet->ref_count++;
+ return packet->ref_count;
+}
+
+int
+sdkctl_packet_release(SDKCtlPacket* packet)
+{
+ assert(packet->ref_count > 0);
+ packet->ref_count--;
+ if (packet->ref_count == 0) {
+ /* Last reference has been dropped. Destroy this object. */
+ _sdkctl_packet_free(packet);
+ return 0;
+ }
+ return packet->ref_count;
+}
+
+/********************************************************************************
+ * SDKCtlQuery implementation
+ *******************************************************************************/
+
+/* Frees query descriptor. */
+static void
+_sdkctl_query_free(SDKCtlQuery* query)
+{
+ if (query != NULL) {
+ SDKCtlSocket* const sdkctl = query->sdkctl;
+ T("SDKCtl %s: Query %p ID %d is freed.",
+ sdkctl->service_name, query, query->header.query_id);
+
+ /* Free allocated resources. */
+ if (query->internal_resp_buffer != NULL &&
+ (query->response_buffer == NULL ||
+ query->response_buffer == &query->internal_resp_buffer)) {
+ free(query->internal_resp_buffer);
+ }
+
+ loopTimer_done(query->timer);
+ _sdkctl_socket_free_recycler(sdkctl, query);
+
+ /* Release socket that owned this query. */
+ sdkctl_socket_release(sdkctl);
+ }
+}
+
+/* Cancels timeout for the query.
+ *
+ * For the simplicity of implementation, the dispatcher will cancel query timer
+ * when query response data begins to flow in. If we let the timer to expire at
+ * that stage, we will end up with data flowing in without real place to
+ * accomodate it.
+ */
+static void
+_sdkctl_query_cancel_timeout(SDKCtlQuery* query)
+{
+ loopTimer_stop(query->timer);
+
+ T("SDKCtl %s: Query %p ID %d deadline is cancelled.",
+ query->sdkctl->service_name, query, query->header.query_id);
+}
+
+/*
+ * Query I/O callbacks.
+ */
+
+/* Callback that is invoked by the I/O dispatcher when query is successfuly
+ * completed (i.e. response to the query is received).
+ */
+static void
+_on_sdkctl_query_completed(SDKCtlQuery* query)
+{
+ T("SDKCtl %s: Query %p ID %d is completed.",
+ query->sdkctl->service_name, query, query->header.query_id);
+
+ /* Cancel deadline, and inform the client about query completion. */
+ _sdkctl_query_cancel_timeout(query);
+ query->query_cb(query->query_opaque, query, ASIO_STATE_SUCCEEDED);
+}
+
+/* A callback that is invoked on query cancellation. */
+static void
+_on_sdkctl_query_cancelled(SDKCtlQuery* query)
+{
+ /*
+ * Query cancellation means that SDK controller is disconnected. In turn,
+ * this means that SDK controller socket will handle disconnection in its
+ * connection callback. So, at this point all we need to do here is to inform
+ * the client, and then unlist the query.
+ */
+
+ /* Cancel deadline, and inform the client about query cancellation. */
+ _sdkctl_query_cancel_timeout(query);
+ query->query_cb(query->query_opaque, query, ASIO_STATE_CANCELLED);
+}
+
+/* A timer callback that is invoked on query timeout.
+ * Param:
+ * opaque - SDKCtlQuery instance.
+ */
+static void
+_on_skdctl_query_timeout(void* opaque)
+{
+ SDKCtlQuery* const query = (SDKCtlQuery*)opaque;
+
+ D("SDKCtl %s: Query %p ID %d with deadline %lld has timed out at %lld",
+ query->sdkctl->service_name, query, query->header.query_id,
+ query->deadline, async_socket_deadline(query->sdkctl->as, 0));
+
+ /* Reference the query while we're in this callback. */
+ sdkctl_query_reference(query);
+
+ /* Inform the client about deadline expiration. Note that client may
+ * extend the deadline, and retry the query. */
+ const AsyncIOAction action =
+ query->query_cb(query->query_opaque, query, ASIO_STATE_TIMED_OUT);
+
+ /* For actions other than retry we will destroy the query. */
+ if (action != ASIO_ACTION_RETRY) {
+ _sdkctl_socket_remove_query(query);
+ }
+
+ sdkctl_query_release(query);
+}
+
+/* A callback that is invoked when query has been sent to the SDK controller
+ * service. */
+static void
+_on_sdkctl_query_sent(SDKCtlQuery* query)
+{
+ T("SDKCtl %s: sent %d bytes of query %p ID %d of type %d",
+ query->sdkctl->service_name, query->header.packet.size, query,
+ query->header.query_id, query->header.query_type);
+
+ /* Inform the client about the event. */
+ query->query_cb(query->query_opaque, query, ASIO_STATE_CONTINUES);
+
+ /* Set a timer to expire at query's deadline, and let the response to come
+ * through the dispatcher loop. */
+ loopTimer_startAbsolute(query->timer, query->deadline);
+}
+
+/* An I/O callback invoked on query transmission.
+ * Param:
+ * io_opaque SDKCtlQuery instance of the query that's being sent with this I/O.
+ * asio - Write I/O descriptor.
+ * status - I/O status.
+ */
+static AsyncIOAction
+_on_sdkctl_query_send_io(void* io_opaque,
+ AsyncSocketIO* asio,
+ AsyncIOState status)
+{
+ SDKCtlQuery* const query = (SDKCtlQuery*)io_opaque;
+ AsyncIOAction action = ASIO_ACTION_DONE;
+
+ /* Reference the query while we're in this callback. */
+ sdkctl_query_reference(query);
+
+ if (status == ASIO_STATE_SUCCEEDED) {
+ /* Query has been sent to the service. */
+ _on_sdkctl_query_sent(query);
+
+ sdkctl_query_release(query);
+
+ return ASIO_ACTION_DONE;
+ }
+
+ /* Lets see what's going on with query transmission. */
+ switch (status) {
+ case ASIO_STATE_CANCELLED:
+ T("SDKCtl %s: Query %p ID %d is cancelled in %s I/O.",
+ query->sdkctl->service_name, query, query->header.query_id,
+ async_socket_io_is_read(asio) ? "READ" : "WRITE");
+ /* Remove the query from the list of active queries. */
+ _sdkctl_socket_remove_query(query);
+ _on_sdkctl_query_cancelled(query);
+ break;
+
+ case ASIO_STATE_TIMED_OUT:
+ D("SDKCtl %s: Query %p ID %d with deadline %lld has timed out in %s I/O at %lld",
+ query->sdkctl->service_name, query, query->header.query_id,
+ query->deadline, async_socket_io_is_read(asio) ? "READ" : "WRITE",
+ async_socket_deadline(query->sdkctl->as, 0));
+ /* Invoke query's callback. */
+ action = query->query_cb(query->query_opaque, query, status);
+ /* For actions other than retry we need to stop the query. */
+ if (action != ASIO_ACTION_RETRY) {
+ _sdkctl_socket_remove_query(query);
+ }
+ break;
+
+ case ASIO_STATE_FAILED:
+ T("SDKCtl %s: Query %p ID %d failed in %s I/O: %d -> %s",
+ query->sdkctl->service_name, query, query->header.query_id,
+ async_socket_io_is_read(asio) ? "READ" : "WRITE",
+ errno, strerror(errno));
+ /* Invoke query's callback. Note that we will let the client to
+ * decide what to do on I/O failure. */
+ action = query->query_cb(query->query_opaque, query, status);
+ /* For actions other than retry we need to stop the query. */
+ if (action != ASIO_ACTION_RETRY) {
+ _sdkctl_socket_remove_query(query);
+ }
+ break;
+
+ case ASIO_STATE_FINISHED:
+ /* Time to disassociate with the I/O. */
+ sdkctl_query_release(query);
+ break;
+
+ default:
+ /* Transitional state. */
+ break;
+ }
+
+ sdkctl_query_release(query);
+
+ return action;
+}
+
+/********************************************************************************
+ * SDKCtlQuery public API implementation
+ ********************************************************************************/
+
+SDKCtlQuery*
+sdkctl_query_new(SDKCtlSocket* sdkctl, int query_type, uint32_t in_data_size)
+{
+ const uint32_t total_size = sizeof(SDKCtlQuery) + in_data_size;
+
+ SDKCtlQuery* const query = _sdkctl_socket_alloc_recycler(sdkctl, total_size);
+ query->next = NULL;
+ query->sdkctl = sdkctl;
+ query->response_buffer = NULL;
+ query->response_size = NULL;
+ query->internal_resp_buffer = NULL;
+ query->internal_resp_size = 0;
+ query->query_cb = NULL;
+ query->query_opaque = NULL;
+ query->deadline = DURATION_INFINITE;
+ query->ref_count = 1;
+ query->header.packet.size = sizeof(SDKCtlQueryHeader) + in_data_size;
+ query->header.packet.type = SDKCTL_PACKET_QUERY;
+ query->header.query_id = _sdkctl_socket_next_query_id(sdkctl);
+ query->header.query_type = query_type;
+
+ /* Initialize timer to fire up on query deadline expiration. */
+ loopTimer_init(query->timer, sdkctl->looper, _on_skdctl_query_timeout, query);
+
+ /* Reference socket that owns this query. */
+ sdkctl_socket_reference(sdkctl);
+
+ T("SDKCtl %s: Query %p ID %d type %d is created for %d bytes of data.",
+ query->sdkctl->service_name, query, query->header.query_id,
+ query_type, in_data_size);
+
+ return query;
+}
+
+SDKCtlQuery*
+sdkctl_query_new_ex(SDKCtlSocket* sdkctl,
+ int query_type,
+ uint32_t in_data_size,
+ const void* in_data,
+ void** response_buffer,
+ uint32_t* response_size,
+ on_sdkctl_query_cb query_cb,
+ void* query_opaque)
+{
+ SDKCtlQuery* const query = sdkctl_query_new(sdkctl, query_type, in_data_size);
+
+ query->response_buffer = response_buffer;
+ if (query->response_buffer == NULL) {
+ /* Creator didn't supply a buffer. Use internal one instead. */
+ query->response_buffer = &query->internal_resp_buffer;
+ query->internal_resp_buffer = NULL;
+ }
+ query->response_size = response_size;
+ if (query->response_size == NULL) {
+ /* Creator didn't supply a buffer for response size. Use internal one
+ * instead. */
+ query->response_size = &query->internal_resp_size;
+ query->internal_resp_size = 0;
+ }
+ query->query_cb = query_cb;
+ query->query_opaque = query_opaque;
+ /* Init query's input buffer. */
+ if (in_data_size != 0 && in_data != NULL) {
+ memcpy(query + 1, in_data, in_data_size);
+ }
+
+ return query;
+}
+
+void
+sdkctl_query_send(SDKCtlQuery* query, int to)
+{
+ SDKCtlSocket* const sdkctl = query->sdkctl;
+
+ /* Initialize the deadline. */
+ query->deadline = async_socket_deadline(query->sdkctl->as, to);
+
+ /* List the query in the list of active queries. */
+ _sdkctl_socket_add_query(query);
+
+ /* Reference query associated with write I/O. */
+ sdkctl_query_reference(query);
+
+ /* Transmit the query to SDK controller. */
+ async_socket_write_abs(sdkctl->as, &query->header, query->header.packet.size,
+ _on_sdkctl_query_send_io, query, query->deadline);
+
+ T("SDKCtl %s: Query %p ID %d type %d is sent with deadline at %lld",
+ query->sdkctl->service_name, query, query->header.query_id,
+ query->header.query_type, query->deadline);
+}
+
+SDKCtlQuery*
+sdkctl_query_build_and_send(SDKCtlSocket* sdkctl,
+ int query_type,
+ uint32_t in_data_size,
+ const void* in_data,
+ void** response_buffer,
+ uint32_t* response_size,
+ on_sdkctl_query_cb query_cb,
+ void* query_opaque,
+ int to)
+{
+ SDKCtlQuery* const query =
+ sdkctl_query_new_ex(sdkctl, query_type, in_data_size, in_data,
+ response_buffer, response_size, query_cb,
+ query_opaque);
+ sdkctl_query_send(query, to);
+ return query;
+}
+
+int
+sdkctl_query_reference(SDKCtlQuery* query)
+{
+ assert(query->ref_count > 0);
+ query->ref_count++;
+ return query->ref_count;
+}
+
+int
+sdkctl_query_release(SDKCtlQuery* query)
+{
+ assert(query->ref_count > 0);
+ query->ref_count--;
+ if (query->ref_count == 0) {
+ /* Last reference has been dropped. Destroy this object. */
+ _sdkctl_query_free(query);
+ return 0;
+ }
+ return query->ref_count;
+}
+
+/********************************************************************************
+ * SDKCtlPacket implementation
+ *******************************************************************************/
+
+/* A packet has been received from SDK controller. */
+static void
+_on_sdkctl_packet_received(SDKCtlSocket* sdkctl, SDKCtlPacket* packet)
+{
+ T("SDKCtl %s: Received packet size: %d, type: %d",
+ sdkctl->service_name, packet->header.size, packet->header.type);
+
+ /* Dispatch received packet to the client. */
+ sdkctl->on_message(sdkctl->opaque, sdkctl, packet, packet->header.type,
+ packet + 1, packet->header.size - sizeof(SDKCtlPacketHeader));
+}
+
+/********************************************************************************
+ * SDKCtlIODispatcher implementation
+ *******************************************************************************/
+
+/* An I/O callback invoked when data gets received from the socket.
+ * Param:
+ * io_opaque SDKCtlIODispatcher instance associated with the reader.
+ * asio - Read I/O descriptor.
+ * status - I/O status.
+ */
+static AsyncIOAction _on_sdkctl_io_dispatcher_io(void* io_opaque,
+ AsyncSocketIO* asio,
+ AsyncIOState status);
+
+/* Starts I/O dispatcher for SDK controller socket. */
+static void
+_sdkctl_io_dispatcher_start(SDKCtlSocket* sdkctl) {
+ SDKCtlIODispatcher* const dispatcher = &sdkctl->io_dispatcher;
+
+ dispatcher->state = SDKCTL_IODISP_EXPECT_HEADER;
+ dispatcher->sdkctl = sdkctl;
+ dispatcher->packet = NULL;
+ dispatcher->current_query = NULL;
+
+ /* Register a packet header reader with the socket. */
+ async_socket_read_rel(dispatcher->sdkctl->as, &dispatcher->header,
+ sizeof(SDKCtlPacketHeader), _on_sdkctl_io_dispatcher_io,
+ dispatcher, -1);
+}
+
+/* Resets I/O dispatcher for SDK controller socket. */
+static void
+_sdkctl_io_dispatcher_reset(SDKCtlSocket* sdkctl) {
+ SDKCtlIODispatcher* const dispatcher = &sdkctl->io_dispatcher;
+
+ /* Cancel current query. */
+ if (dispatcher->current_query != NULL) {
+ SDKCtlQuery* const query = dispatcher->current_query;
+ dispatcher->current_query = NULL;
+ _on_sdkctl_query_cancelled(query);
+ sdkctl_query_release(query);
+ }
+
+ /* Free packet data buffer. */
+ if (dispatcher->packet != NULL) {
+ sdkctl_packet_release(dispatcher->packet);
+ dispatcher->packet = NULL;
+ }
+
+ /* Reset dispatcher state. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_HEADER;
+}
+
+/*
+ * I/O dispatcher callbacks.
+ */
+
+/* A callback that is invoked when a failure occurred while dispatcher was
+ * reading data from the socket.
+ */
+static void
+_on_io_dispatcher_io_failure(SDKCtlIODispatcher* dispatcher,
+ AsyncSocketIO* asio)
+{
+ SDKCtlSocket* const sdkctl = dispatcher->sdkctl;
+
+ D("SDKCtl %s: Dispatcher I/O failure: %d -> %s",
+ sdkctl->service_name, errno, strerror(errno));
+
+ /* We treat all I/O failures same way we treat disconnection. Just cancel
+ * everything, disconnect, and let the client to decide what to do next. */
+ sdkctl_socket_disconnect(sdkctl);
+
+ /* Report disconnection to the client, and let it restore connection in this
+ * callback. */
+ sdkctl->on_connection(sdkctl->opaque, sdkctl, ASIO_STATE_FAILED);
+}
+
+/* A callback that is invoked when dispatcher's reader has been cancelled. */
+static void
+_on_io_dispatcher_io_cancelled(SDKCtlIODispatcher* dispatcher,
+ AsyncSocketIO* asio)
+{
+ T("SDKCtl %s: Dispatcher I/O cancelled.", dispatcher->sdkctl->service_name);
+
+ /* If we're in the middle of receiving query reply we need to cancel the
+ * query. */
+ if (dispatcher->current_query != NULL) {
+ SDKCtlQuery* const query = dispatcher->current_query;
+ dispatcher->current_query = NULL;
+ _on_sdkctl_query_cancelled(query);
+ sdkctl_query_release(query);
+ }
+
+ /* Discard packet data we've received so far. */
+ if (dispatcher->packet != NULL) {
+ sdkctl_packet_release(dispatcher->packet);
+ dispatcher->packet = NULL;
+ }
+}
+
+/* A generic packet header has been received by I/O dispatcher. */
+static AsyncIOAction
+_on_io_dispatcher_packet_header(SDKCtlIODispatcher* dispatcher,
+ AsyncSocketIO* asio)
+{
+ SDKCtlSocket* const sdkctl = dispatcher->sdkctl;
+
+ T("SDKCtl %s: Packet header type %d, size %d is received.",
+ dispatcher->sdkctl->service_name, dispatcher->header.type,
+ dispatcher->header.size);
+
+ /* Here we have three choices for the packet, that define the rest of
+ * the data that follow it:
+ * - Regular packet,
+ * - Response to a query that has been sent to SDK controller,
+ * - A query from SDK controller.
+ * Update the state accordingly, and initiate reading of the
+ * remaining of the packet.
+ */
+ if (dispatcher->header.type == SDKCTL_PACKET_QUERY_RESPONSE) {
+ /* This is a response to the query. Before receiving response data we
+ * need to locate the relevant query, and use its response buffer to read
+ * the data. For that we need to obtain query ID firts. So, initiate
+ * reading of the remaining part of SDKCtlQueryReplyHeader. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_QUERY_REPLY_HEADER;
+ async_socket_read_rel(sdkctl->as, &dispatcher->query_reply_header.query_id,
+ sizeof(SDKCtlQueryReplyHeader) - sizeof(SDKCtlPacketHeader),
+ _on_sdkctl_io_dispatcher_io, dispatcher, -1);
+ } else {
+ /* For regular packets, as well as queries, we simply allocate buffer,
+ * that fits the entire packet, and read the remainder of the data in
+ * there. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_DATA;
+ dispatcher->packet =
+ _sdkctl_packet_new(sdkctl, dispatcher->header.size,
+ dispatcher->header.type);
+ /* Initiate reading of the packet data. */
+ async_socket_read_rel(sdkctl->as, dispatcher->packet + 1,
+ dispatcher->header.size - sizeof(SDKCtlPacketHeader),
+ _on_sdkctl_io_dispatcher_io, dispatcher, -1);
+ }
+
+ return ASIO_ACTION_DONE;
+}
+
+/* A generic packet has been received by I/O dispatcher. */
+static AsyncIOAction
+_on_io_dispatcher_packet(SDKCtlIODispatcher* dispatcher, AsyncSocketIO* asio)
+{
+ SDKCtlSocket* const sdkctl = dispatcher->sdkctl;
+
+ T("SDKCtl %s: Packet type %d, size %d is received.",
+ dispatcher->sdkctl->service_name, dispatcher->header.type,
+ dispatcher->header.size);
+
+ _on_sdkctl_packet_received(sdkctl, dispatcher->packet);
+ sdkctl_packet_release(dispatcher->packet);
+ dispatcher->packet = NULL;
+
+ /* Get ready for the next I/O cycle. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_HEADER;
+ async_socket_read_rel(sdkctl->as, &dispatcher->header, sizeof(SDKCtlPacketHeader),
+ _on_sdkctl_io_dispatcher_io, dispatcher, -1);
+ return ASIO_ACTION_DONE;
+}
+
+/* A query reply header has been received by I/O dispatcher. */
+static AsyncIOAction
+_on_io_dispatcher_query_reply_header(SDKCtlIODispatcher* dispatcher,
+ AsyncSocketIO* asio)
+{
+ SDKCtlSocket* const sdkctl = dispatcher->sdkctl;
+ SDKCtlQuery* query;
+
+ T("SDKCtl %s: Query reply header is received for query ID %d",
+ dispatcher->sdkctl->service_name, dispatcher->query_reply_header.query_id);
+
+ /* Pull the query out of the list of active queries. It's the dispatcher that
+ * owns this query now. */
+ dispatcher->current_query =
+ _sdkctl_socket_remove_query_id(sdkctl, dispatcher->query_reply_header.query_id);
+ query = dispatcher->current_query;
+
+ if (query == NULL) {
+ D("%s: Query #%d is not found by dispatcher",
+ dispatcher->sdkctl->service_name, dispatcher->query_reply_header.query_id);
+
+ /* Query is not found. Just read the remainder of reply up in the air,
+ * and then discard when it's over. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_QUERY_REPLY_DATA;
+ dispatcher->packet =
+ _sdkctl_packet_new(sdkctl, dispatcher->header.size,
+ dispatcher->header.type);
+ /* Copy query reply info to the packet. */
+ memcpy(&dispatcher->packet->header, &dispatcher->query_reply_header,
+ sizeof(SDKCtlQueryReplyHeader));
+ async_socket_read_rel(sdkctl->as, &dispatcher->query_header + 1,
+ dispatcher->header.size - sizeof(SDKCtlQueryReplyHeader),
+ _on_sdkctl_io_dispatcher_io, dispatcher, -1);
+ } else {
+ /* Prepare to receive query reply. For the simplicity sake, cancel query
+ * time out, so it doesn't expire on us while we're in the middle of
+ * receiving query's reply. */
+ _sdkctl_query_cancel_timeout(query);
+
+ /* Adjust the reply buffer set for the query (if needed). */
+ const uint32_t query_data_size =
+ dispatcher->header.size - sizeof(SDKCtlQueryReplyHeader);
+ if (*query->response_size < query_data_size) {
+ *query->response_buffer = malloc(query_data_size);
+ if (*query->response_buffer == NULL) {
+ APANIC("%s: Unable to allocate %d bytes for query response",
+ sdkctl->service_name, query_data_size);
+ }
+ }
+ /* Save the actual query response size. */
+ *query->response_size = query_data_size;
+
+ /* Start reading query response. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_QUERY_REPLY_DATA;
+ async_socket_read_rel(sdkctl->as, *query->response_buffer,
+ *query->response_size, _on_sdkctl_io_dispatcher_io,
+ dispatcher, -1);
+ }
+
+ return ASIO_ACTION_DONE;
+}
+
+/* A query reply header has been received by I/O dispatcher. */
+static AsyncIOAction
+_on_io_dispatcher_query_reply(SDKCtlIODispatcher* dispatcher, AsyncSocketIO* asio)
+{
+ SDKCtlSocket* const sdkctl = dispatcher->sdkctl;
+ SDKCtlQuery* const query = dispatcher->current_query;
+ dispatcher->current_query = NULL;
+
+ if (query != NULL) {
+ _ANDROID_ASSERT(query->header.query_id == dispatcher->query_reply_header.query_id,
+ "SDKCtl %s: Query ID mismatch in I/O dispatcher",
+ sdkctl->service_name);
+ T("SDKCtl %s: Query reply is received for query %p ID %d. Reply size is %d",
+ dispatcher->sdkctl->service_name, query, query->header.query_id,
+ *query->response_size);
+
+ /* Complete the query, and release it from the dispatcher. */
+ _on_sdkctl_query_completed(query);
+ sdkctl_query_release(query);
+ } else {
+ /* This was "read up in the air" for a cancelled query. Just discard the
+ * read data. */
+ if (dispatcher->packet != NULL) {
+ sdkctl_packet_release(dispatcher->packet);
+ dispatcher->packet = NULL;
+ }
+ }
+
+ /* Get ready for the next I/O cycle. */
+ dispatcher->state = SDKCTL_IODISP_EXPECT_HEADER;
+ async_socket_read_rel(sdkctl->as, &dispatcher->header, sizeof(SDKCtlPacketHeader),
+ _on_sdkctl_io_dispatcher_io, dispatcher, -1);
+ return ASIO_ACTION_DONE;
+}
+
+/* An I/O callback invoked when data gets received from the socket.
+ * This is main I/O dispatcher loop.
+ * Param:
+ * io_opaque SDKCtlIODispatcher instance associated with the reader.
+ * asio - Read I/O descriptor.
+ * status - I/O status.
+ */
+static AsyncIOAction
+_on_sdkctl_io_dispatcher_io(void* io_opaque,
+ AsyncSocketIO* asio,
+ AsyncIOState status)
+{
+ AsyncIOAction action = ASIO_ACTION_DONE;
+ SDKCtlIODispatcher* const dispatcher = (SDKCtlIODispatcher*)io_opaque;
+ SDKCtlSocket* const sdkctl = dispatcher->sdkctl;
+
+ /* Reference SDKCtlSocket while we're in this callback. */
+ sdkctl_socket_reference(sdkctl);
+
+ if (status != ASIO_STATE_SUCCEEDED) {
+ /* Something going on with I/O other than receiving data.. */
+ switch (status) {
+ case ASIO_STATE_STARTED:
+ /* Data has started flowing in. Cancel timeout on I/O that has
+ * started, so we can complete the current state of the
+ * dispatcher without interruptions other than I/O failures. */
+ async_socket_io_cancel_time_out(asio);
+ break;
+
+ case ASIO_STATE_FAILED:
+ /* I/O failure has occurred. Handle the failure. */
+ _on_io_dispatcher_io_failure(dispatcher, asio);
+ break;
+
+ case ASIO_STATE_TIMED_OUT:
+ /* The way I/O dispatcher is implemented, this should never
+ * happen, because dispatcher doesn't set I/O expiration time
+ * when registering its readers. */
+ _ANDROID_ASSERT(0,
+ "SDKCtl %s: We should never receive ASIO_STATE_TIMED_OUT in SDKCtl I/O dispatcher.",
+ sdkctl->service_name);
+ break;
+
+ case ASIO_STATE_CANCELLED:
+ /* Cancellation means that we're in the middle of handling
+ * disconnection. Sooner or later, this dispatcher will be reset,
+ * so we don't really care about keeping its state at this point.
+ */
+ _on_io_dispatcher_io_cancelled(dispatcher, asio);
+ break;
+
+ case ASIO_STATE_FINISHED:
+ break;
+
+ default:
+ _ANDROID_ASSERT(0, "SDKCtl %s: Unexpected I/O status %d in the dispatcher",
+ sdkctl->service_name, status);
+ /* Handle this as protocol failure. */
+ errno = EINVAL;
+ _on_io_dispatcher_io_failure(dispatcher, asio);
+ action = ASIO_ACTION_ABORT;
+ break;
+ }
+
+ sdkctl_socket_release(sdkctl);
+
+ return action;
+ }
+
+ /* Requested data has been read. Handle the chunk depending on dispatcher's
+ * state. */
+ switch (dispatcher->state) {
+ case SDKCTL_IODISP_EXPECT_HEADER:
+ /* A generic packet header is received. */
+ action = _on_io_dispatcher_packet_header(dispatcher, asio);
+ break;
+
+ case SDKCTL_IODISP_EXPECT_QUERY_REPLY_HEADER:
+ /* Query reply header is received. */
+ action = _on_io_dispatcher_query_reply_header(dispatcher, asio);
+ break;
+
+ case SDKCTL_IODISP_EXPECT_QUERY_REPLY_DATA:
+ /* Query reply is received. Complete the query. */
+ action = _on_io_dispatcher_query_reply(dispatcher, asio);
+ break;
+
+ case SDKCTL_IODISP_EXPECT_DATA:
+ /* A generic packet is received. */
+ action = _on_io_dispatcher_packet(dispatcher, asio);
+ break;
+
+ default:
+ _ANDROID_ASSERT(0, "SDKCtl %s: Unexpected I/O dispacher state %d",
+ sdkctl->service_name, dispatcher->state);
+ break;
+ }
+
+ sdkctl_socket_release(sdkctl);
+
+ return action;
+}
+
+/********************************************************************************
+ * SDKCtlSocket internals.
+ *******************************************************************************/
+
+/* Cancels all queries that is active on this socket. */
+static void
+_sdkctl_socket_cancel_all_queries(SDKCtlSocket* sdkctl)
+{
+ SDKCtlIODispatcher* const dispatcher = &sdkctl->io_dispatcher;
+ SDKCtlQuery* query;
+
+ /* Cancel query that is being completed in dispatcher. */
+ if (dispatcher->current_query != NULL) {
+ SDKCtlQuery* const query = dispatcher->current_query;
+ dispatcher->current_query = NULL;
+ _on_sdkctl_query_cancelled(query);
+ sdkctl_query_release(query);
+ }
+
+ /* One by one empty query list cancelling pulled queries. */
+ query = _sdkctl_socket_pull_first_query(sdkctl);
+ while (query != NULL) {
+ _sdkctl_query_cancel_timeout(query);
+ query->query_cb(query->query_opaque, query, ASIO_STATE_CANCELLED);
+ sdkctl_query_release(query);
+ query = _sdkctl_socket_pull_first_query(sdkctl);
+ }
+}
+
+/* Cancels all packets that is active on this socket. */
+static void
+_sdkctl_socket_cancel_all_packets(SDKCtlSocket* sdkctl)
+{
+}
+
+/* Cancels all I/O that is active on this socket. */
+static void
+_sdkctl_socket_cancel_all_io(SDKCtlSocket* sdkctl)
+{
+ /* Cancel all queries, and packets that are active for this I/O. */
+ _sdkctl_socket_cancel_all_queries(sdkctl);
+ _sdkctl_socket_cancel_all_packets(sdkctl);
+}
+
+/* Disconnects AsyncSocket for SDKCtlSocket. */
+static void
+_sdkctl_socket_disconnect_socket(SDKCtlSocket* sdkctl)
+{
+ if (sdkctl->as != NULL) {
+ /* Disconnect the socket. This will trigger I/O cancellation callbacks. */
+ async_socket_disconnect(sdkctl->as);
+
+ /* Cancel all I/O that is active on this socket. */
+ _sdkctl_socket_cancel_all_io(sdkctl);
+
+ /* Reset I/O dispatcher. */
+ _sdkctl_io_dispatcher_reset(sdkctl);
+ }
+
+ sdkctl->state = SDKCTL_SOCKET_DISCONNECTED;
+}
+
+/* Frees SDKCtlSocket instance. */
+static void
+_sdkctl_socket_free(SDKCtlSocket* sdkctl)
+{
+ if (sdkctl != NULL) {
+ /* Disconnect, and release the socket. */
+ if (sdkctl->as != NULL) {
+ async_socket_disconnect(sdkctl->as);
+ async_socket_release(sdkctl->as);
+ }
+
+ /* Free allocated resources. */
+ if (sdkctl->looper != NULL) {
+ looper_free(sdkctl->looper);
+ }
+ if (sdkctl->service_name != NULL) {
+ free(sdkctl->service_name);
+ }
+ _sdkctl_socket_empty_recycler(sdkctl);
+
+ AFREE(sdkctl);
+ }
+}
+
+/********************************************************************************
+ * SDK Control Socket connection callbacks.
+ *******************************************************************************/
+
+/* Initiates handshake query when SDK controller socket is connected. */
+static void _sdkctl_do_handshake(SDKCtlSocket* sdkctl);
+
+/* A socket connection is established.
+ * Here we will start I/O dispatcher, and will initiate a handshake with
+ * the SdkController service for this socket. */
+static AsyncIOAction
+_on_async_socket_connected(SDKCtlSocket* sdkctl)
+{
+ D("SDKCtl %s: Socket is connected.", sdkctl->service_name);
+
+ /* Notify the client that connection is established. */
+ const AsyncIOAction action =
+ sdkctl->on_connection(sdkctl->opaque, sdkctl, ASIO_STATE_SUCCEEDED);
+
+ if (action == ASIO_ACTION_DONE) {
+ /* Initialize, and start main I/O dispatcher. */
+ _sdkctl_io_dispatcher_start(sdkctl);
+
+ /* Initiate handshake. */
+ _sdkctl_do_handshake(sdkctl);
+
+ return action;
+ } else {
+ /* Client didn't like something about this connection. */
+ return action;
+ }
+}
+
+/* Handles lost connection with SdkController service. */
+static AsyncIOAction
+_on_async_socket_disconnected(SDKCtlSocket* sdkctl)
+{
+ D("SDKCtl %s: Socket has been disconnected.", sdkctl->service_name);
+
+ _sdkctl_socket_disconnect_socket(sdkctl);
+
+ AsyncIOAction action = sdkctl->on_connection(sdkctl->opaque, sdkctl,
+ ASIO_STATE_FAILED);
+ if (action == ASIO_ACTION_DONE) {
+ /* Default action for disconnect is to reestablish the connection. */
+ action = ASIO_ACTION_RETRY;
+ }
+ if (action == ASIO_ACTION_RETRY) {
+ sdkctl->state = SDKCTL_SOCKET_CONNECTING;
+ }
+ return action;
+}
+
+/* An entry point for all socket connection events.
+ * Here we will dispatch connection events to appropriate handlers.
+ * Param:
+ * client_opaque - SDKCtlSocket isntance.
+ */
+static AsyncIOAction
+_on_async_socket_connection(void* client_opaque,
+ AsyncSocket* as,
+ AsyncIOState status)
+{
+ AsyncIOAction action = ASIO_ACTION_DONE;
+ SDKCtlSocket* const sdkctl = (SDKCtlSocket*)client_opaque;
+
+ /* Reference the socket while in this callback. */
+ sdkctl_socket_reference(sdkctl);
+
+ switch (status) {
+ case ASIO_STATE_SUCCEEDED:
+ sdkctl->state = SDKCTL_SOCKET_CONNECTED;
+ _on_async_socket_connected(sdkctl);
+ break;
+
+ case ASIO_STATE_FAILED:
+ if (sdkctl->state == SDKCTL_SOCKET_CONNECTED) {
+ /* This is disconnection condition. */
+ action = _on_async_socket_disconnected(sdkctl);
+ } else {
+ /* An error has occurred while attempting to connect to socket.
+ * Lets try again... */
+ action = ASIO_ACTION_RETRY;
+ }
+ break;
+
+ case ASIO_STATE_RETRYING:
+ default:
+ action = ASIO_ACTION_RETRY;
+ break;
+ }
+
+ sdkctl_socket_release(sdkctl);
+
+ return action;
+}
+
+/********************************************************************************
+ * SDK Control Socket public API
+ *******************************************************************************/
+
+SDKCtlSocket*
+sdkctl_socket_new(int reconnect_to,
+ const char* service_name,
+ on_sdkctl_connection_cb on_connection,
+ on_sdkctl_handshake_cb on_handshake,
+ on_sdkctl_message_cb on_message,
+ void* opaque)
+{
+ SDKCtlSocket* sdkctl;
+ ANEW0(sdkctl);
+
+ sdkctl->state = SDKCTL_SOCKET_DISCONNECTED;
+ sdkctl->opaque = opaque;
+ sdkctl->service_name = ASTRDUP(service_name);
+ sdkctl->on_connection = on_connection;
+ sdkctl->on_handshake = on_handshake;
+ sdkctl->on_message = on_message;
+ sdkctl->reconnect_to = reconnect_to;
+ sdkctl->as = NULL;
+ sdkctl->next_query_id = 0;
+ sdkctl->query_head = sdkctl->query_tail = NULL;
+ sdkctl->ref_count = 1;
+ sdkctl->recycler = NULL;
+ sdkctl->recycler_block_size = 0;
+ sdkctl->recycler_max = 0;
+ sdkctl->recycler_count = 0;
+
+ sdkctl->looper = looper_newCore();
+ if (sdkctl->looper == NULL) {
+ E("Unable to create I/O looper for SDKCtl socket '%s'",
+ service_name);
+ on_connection(opaque, sdkctl, ASIO_STATE_FAILED);
+ _sdkctl_socket_free(sdkctl);
+ return NULL;
+ }
+
+ return sdkctl;
+}
+
+int sdkctl_socket_reference(SDKCtlSocket* sdkctl)
+{
+ assert(sdkctl->ref_count > 0);
+ sdkctl->ref_count++;
+ return sdkctl->ref_count;
+}
+
+int
+sdkctl_socket_release(SDKCtlSocket* sdkctl)
+{
+ assert(sdkctl->ref_count > 0);
+ sdkctl->ref_count--;
+ if (sdkctl->ref_count == 0) {
+ /* Last reference has been dropped. Destroy this object. */
+ _sdkctl_socket_free(sdkctl);
+ return 0;
+ }
+ return sdkctl->ref_count;
+}
+
+void
+sdkctl_init_recycler(SDKCtlSocket* sdkctl,
+ uint32_t data_size,
+ int max_recycled_num)
+{
+ if (sdkctl->recycler != NULL) {
+ D("SDKCtl %s: Recycler is already initialized. Ignoring recycler init.",
+ sdkctl->service_name);
+ return;
+ }
+
+ /* SDKCtlQuery is max descriptor sizeof. */
+ data_size += sizeof(SDKCtlQuery);
+
+ sdkctl->recycler_block_size = data_size;
+ sdkctl->recycler_max = max_recycled_num;
+ sdkctl->recycler_count = 0;
+}
+
+void
+sdkctl_socket_connect(SDKCtlSocket* sdkctl, int port, int retry_to)
+{
+ T("SDKCtl %s: Handling connect request to port %d, retrying in %dms...",
+ sdkctl->service_name, port, retry_to);
+
+ sdkctl->state = SDKCTL_SOCKET_CONNECTING;
+ sdkctl->as = async_socket_new(port, sdkctl->reconnect_to,
+ _on_async_socket_connection, sdkctl,
+ sdkctl->looper);
+ if (sdkctl->as == NULL) {
+ E("Unable to allocate AsyncSocket for SDKCtl socket '%s'",
+ sdkctl->service_name);
+ sdkctl->on_connection(sdkctl->opaque, sdkctl, ASIO_STATE_FAILED);
+ } else {
+ async_socket_connect(sdkctl->as, retry_to);
+ }
+}
+
+void
+sdkctl_socket_reconnect(SDKCtlSocket* sdkctl, int port, int retry_to)
+{
+ T("SDKCtl %s: Handling reconnection request to port %d, retrying in %dms...",
+ sdkctl->service_name, port, retry_to);
+
+ _sdkctl_socket_disconnect_socket(sdkctl);
+
+ if (sdkctl->as == NULL) {
+ sdkctl_socket_connect(sdkctl, port, retry_to);
+ } else {
+ sdkctl->state = SDKCTL_SOCKET_CONNECTING;
+ async_socket_reconnect(sdkctl->as, retry_to);
+ }
+}
+
+void
+sdkctl_socket_disconnect(SDKCtlSocket* sdkctl)
+{
+ T("SDKCtl %s: Handling disconnect request.", sdkctl->service_name);
+
+ _sdkctl_socket_disconnect_socket(sdkctl);
+}
+
+
+/********************************************************************************
+ * Handshake query
+ *******************************************************************************/
+
+/* A callback that is ivoked on handshake I/O events. */
+static AsyncIOAction
+_on_handshake_io(void* query_opaque,
+ SDKCtlQuery* query,
+ AsyncIOState status)
+{
+ SDKCtlSocket* const sdkctl = (SDKCtlSocket*)query_opaque;
+
+ if (status == ASIO_STATE_SUCCEEDED) {
+ D("SDKCtl %s: %d bytes of handshake reply is received.",
+ sdkctl->service_name, *query->response_size);
+
+ /* Handshake is received. Inform the client. */
+ sdkctl->on_handshake(sdkctl->opaque, sdkctl, *query->response_buffer,
+ *query->response_size, status);
+ } else {
+ /* Something is going on with the handshake... */
+ switch (status) {
+ case ASIO_STATE_FAILED:
+ case ASIO_STATE_TIMED_OUT:
+ case ASIO_STATE_CANCELLED:
+ D("SDKCtl %s: Handshake failed: I/O state %d. Error: %d -> %s",
+ sdkctl->service_name, status, errno, strerror(errno));
+ sdkctl->on_handshake(sdkctl->opaque, sdkctl,
+ *query->response_buffer,
+ *query->response_size, status);
+ break;
+
+ default:
+ break;
+ }
+ }
+ return ASIO_ACTION_DONE;
+}
+
+static AsyncIOAction
+_on_sdkctl_endianness_io(void* io_opaque,
+ AsyncSocketIO* asio,
+ AsyncIOState status) {
+ SDKCtlSocket* const sdkctl = (SDKCtlSocket*)io_opaque;
+
+ if (status == ASIO_STATE_SUCCEEDED) {
+ /* Now it's time to initiate handshake message. */
+ D("SDKCtl %s: Sending handshake query...", sdkctl->service_name);
+ SDKCtlQuery* query =
+ sdkctl_query_build_and_send(sdkctl, SDKCTL_QUERY_HANDSHAKE,
+ strlen(sdkctl->service_name),
+ sdkctl->service_name, NULL, NULL,
+ _on_handshake_io, sdkctl, 3000);
+ sdkctl_query_release(query);
+ return ASIO_ACTION_DONE;
+ } else {
+ /* Something is going on with the endianness... */
+ switch (status) {
+ case ASIO_STATE_FAILED:
+ case ASIO_STATE_TIMED_OUT:
+ case ASIO_STATE_CANCELLED:
+ D("SDKCtl %s: endianness failed: I/O state %d. Error: %d -> %s",
+ sdkctl->service_name, status, errno, strerror(errno));
+ sdkctl->on_handshake(sdkctl->opaque, sdkctl, NULL, 0, status);
+ break;
+
+ default:
+ break;
+ }
+ }
+ return ASIO_ACTION_DONE;
+}
+
+static void
+_sdkctl_do_handshake(SDKCtlSocket* sdkctl)
+{
+#ifndef HOST_WORDS_BIGENDIAN
+static const char _host_end = 0;
+#else
+static const char _host_end = 1;
+#endif
+
+ D("SDKCtl %s: Sending endianness: %d...", sdkctl->service_name, _host_end);
+
+ /* Before we can send any structured data to the SDK controller we need to
+ * report endianness of the host. */
+ async_socket_write_rel(sdkctl->as, &_host_end, 1,
+ _on_sdkctl_endianness_io, sdkctl, 3000);
+}