/*
* 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 "android/utils/debug.h"
#include "android/async-socket-connector.h"
#include "android/async-socket.h"
#include "utils/panic.h"
#include "iolooper.h"
#define E(...) derror(__VA_ARGS__)
#define W(...) dwarning(__VA_ARGS__)
#define D(...) VERBOSE_PRINT(asyncsocket,__VA_ARGS__)
#define D_ACTIVE VERBOSE_CHECK(asyncsocket)
#define TRACE_ON 0
#if TRACE_ON
#define T(...) VERBOSE_PRINT(asyncsocket,__VA_ARGS__)
#else
#define T(...)
#endif
/********************************************************************************
* Asynchronous Socket internal API declarations
*******************************************************************************/
/* Gets socket's address string. */
static const char* _async_socket_string(AsyncSocket* as);
/* Gets socket's looper. */
static Looper* _async_socket_get_looper(AsyncSocket* as);
/* Handler for the I/O time out.
* Param:
* as - Asynchronous socket for the I/O.
* asio - Desciptor for the timed out I/O.
*/
static AsyncIOAction _async_socket_io_timed_out(AsyncSocket* as,
AsyncSocketIO* asio);
/********************************************************************************
* Asynchronous Socket Reader / Writer
*******************************************************************************/
struct AsyncSocketIO {
/* Next I/O in the reader, or writer list. */
AsyncSocketIO* next;
/* Asynchronous socket for this I/O. */
AsyncSocket* as;
/* Timer used for time outs on this I/O. */
LoopTimer timer[1];
/* An opaque pointer associated with this I/O. */
void* io_opaque;
/* Buffer where to read / write data. */
uint8_t* buffer;
/* Bytes to transfer through the socket for this I/O. */
uint32_t to_transfer;
/* Bytes thransferred through the socket in this I/O. */
uint32_t transferred;
/* I/O callback for this I/O. */
on_as_io_cb on_io;
/* I/O type selector: 1 - read, 0 - write. */
int is_io_read;
/* State of the I/O. */
AsyncIOState state;
/* Number of outstanding references to the I/O. */
int ref_count;
/* Deadline for this I/O */
Duration deadline;
};
/*
* Recycling I/O instances.
* Since AsyncSocketIO instances are not that large, it makes sence to recycle
* them for faster allocation, rather than allocating and freeing them for each
* I/O on the socket.
*/
/* List of recycled I/O descriptors. */
static AsyncSocketIO* _asio_recycled = NULL;
/* Number of I/O descriptors that are recycled in the _asio_recycled list. */
static int _recycled_asio_count = 0;
/* Maximum number of I/O descriptors that can be recycled. */
static const int _max_recycled_asio_num = 32;
/* Handler for an I/O time-out timer event.
* When this routine is invoked, it indicates that a time out has occurred on an
* I/O.
* Param:
* opaque - AsyncSocketIO instance representing the timed out I/O.
*/
static void _on_async_socket_io_timed_out(void* opaque);
/* Creates new I/O descriptor.
* Param:
* as - Asynchronous socket for the I/O.
* is_io_read - I/O type selector: 1 - read, 0 - write.
* buffer, len - Reader / writer buffer address.
* io_cb - Callback for this reader / writer.
* io_opaque - An opaque pointer associated with the I/O.
* deadline - Deadline to complete the I/O.
* Return:
* Initialized AsyncSocketIO instance.
*/
static AsyncSocketIO*
_async_socket_rw_new(AsyncSocket* as,
int is_io_read,
void* buffer,
uint32_t len,
on_as_io_cb io_cb,
void* io_opaque,
Duration deadline)
{
/* Lookup in the recycler first. */
AsyncSocketIO* asio = _asio_recycled;
if (asio != NULL) {
/* Pull the descriptor from recycler. */
_asio_recycled = asio->next;
_recycled_asio_count--;
} else {
/* No recycled descriptors. Allocate new one. */
ANEW0(asio);
}
asio->next = NULL;
asio->as = as;
asio->is_io_read = is_io_read;
asio->buffer = (uint8_t*)buffer;
asio->to_transfer = len;
asio->transferred = 0;
asio->on_io = io_cb;
asio->io_opaque = io_opaque;
asio->state = ASIO_STATE_QUEUED;
asio->ref_count = 1;
asio->deadline = deadline;
loopTimer_init(asio->timer, _async_socket_get_looper(as),
_on_async_socket_io_timed_out, asio);
loopTimer_startAbsolute(asio->timer, deadline);
/* Reference socket that is holding this I/O. */
async_socket_reference(as);
T("ASocket %s: %s I/O descriptor %p is created for %d bytes of data",
_async_socket_string(as), is_io_read ? "READ" : "WRITE", asio, len);
return asio;
}
/* Destroys and frees I/O descriptor. */
static void
_async_socket_io_free(AsyncSocketIO* asio)
{
AsyncSocket* const as = asio->as;
T("ASocket %s: %s I/O descriptor %p is destroyed.",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE", asio);
loopTimer_done(asio->timer);
/* Try to recycle it first, and free the memory if recycler is full. */
if (_recycled_asio_count < _max_recycled_asio_num) {
asio->next = _asio_recycled;
_asio_recycled = asio;
_recycled_asio_count++;
} else {
AFREE(asio);
}
/* Release socket that is holding this I/O. */
async_socket_release(as);
}
/* An I/O has been finished and its descriptor is about to be discarded. */
static void
_async_socket_io_finished(AsyncSocketIO* asio)
{
/* Notify the client of the I/O that I/O is finished. */
asio->on_io(asio->io_opaque, asio, ASIO_STATE_FINISHED);
}
int
async_socket_io_reference(AsyncSocketIO* asio)
{
assert(asio->ref_count > 0);
asio->ref_count++;
return asio->ref_count;
}
int
async_socket_io_release(AsyncSocketIO* asio)
{
assert(asio->ref_count > 0);
asio->ref_count--;
if (asio->ref_count == 0) {
_async_socket_io_finished(asio);
/* Last reference has been dropped. Destroy this object. */
_async_socket_io_free(asio);
return 0;
}
return asio->ref_count;
}
/* Creates new asynchronous socket reader.
* Param:
* as - Asynchronous socket for the reader.
* buffer, len - Reader's buffer.
* io_cb - Reader's callback.
* reader_opaque - An opaque pointer associated with the reader.
* deadline - Deadline to complete the operation.
* Return:
* An initialized AsyncSocketIO intance.
*/
static AsyncSocketIO*
_async_socket_reader_new(AsyncSocket* as,
void* buffer,
uint32_t len,
on_as_io_cb io_cb,
void* reader_opaque,
Duration deadline)
{
AsyncSocketIO* const asio = _async_socket_rw_new(as, 1, buffer, len, io_cb,
reader_opaque, deadline);
return asio;
}
/* Creates new asynchronous socket writer.
* Param:
* as - Asynchronous socket for the writer.
* buffer, len - Writer's buffer.
* io_cb - Writer's callback.
* writer_opaque - An opaque pointer associated with the writer.
* deadline - Deadline to complete the operation.
* Return:
* An initialized AsyncSocketIO intance.
*/
static AsyncSocketIO*
_async_socket_writer_new(AsyncSocket* as,
const void* buffer,
uint32_t len,
on_as_io_cb io_cb,
void* writer_opaque,
Duration deadline)
{
AsyncSocketIO* const asio = _async_socket_rw_new(as, 0, (void*)buffer, len,
io_cb, writer_opaque,
deadline);
return asio;
}
/* I/O timed out. */
static void
_on_async_socket_io_timed_out(void* opaque)
{
AsyncSocketIO* const asio = (AsyncSocketIO*)opaque;
AsyncSocket* const as = asio->as;
D("ASocket %s: %s I/O with deadline %lld has timed out at %lld",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE",
asio->deadline, async_socket_deadline(as, 0));
/* Reference while in callback. */
async_socket_io_reference(asio);
_async_socket_io_timed_out(asio->as, asio);
async_socket_io_release(asio);
}
/********************************************************************************
* Public Asynchronous Socket I/O API
*******************************************************************************/
AsyncSocket*
async_socket_io_get_socket(const AsyncSocketIO* asio)
{
async_socket_reference(asio->as);
return asio->as;
}
void
async_socket_io_cancel_time_out(AsyncSocketIO* asio)
{
loopTimer_stop(asio->timer);
}
void*
async_socket_io_get_io_opaque(const AsyncSocketIO* asio)
{
return asio->io_opaque;
}
void*
async_socket_io_get_client_opaque(const AsyncSocketIO* asio)
{
return async_socket_get_client_opaque(asio->as);
}
void*
async_socket_io_get_buffer_info(const AsyncSocketIO* asio,
uint32_t* transferred,
uint32_t* to_transfer)
{
if (transferred != NULL) {
*transferred = asio->transferred;
}
if (to_transfer != NULL) {
*to_transfer = asio->to_transfer;
}
return asio->buffer;
}
void*
async_socket_io_get_buffer(const AsyncSocketIO* asio)
{
return asio->buffer;
}
uint32_t
async_socket_io_get_transferred(const AsyncSocketIO* asio)
{
return asio->transferred;
}
uint32_t
async_socket_io_get_to_transfer(const AsyncSocketIO* asio)
{
return asio->to_transfer;
}
int
async_socket_io_is_read(const AsyncSocketIO* asio)
{
return asio->is_io_read;
}
/********************************************************************************
* Asynchronous Socket internals
*******************************************************************************/
struct AsyncSocket {
/* TCP address for the socket. */
SockAddress address;
/* Connection callback for this socket. */
on_as_connection_cb on_connection;
/* An opaque pointer associated with this socket by the client. */
void* client_opaque;
/* I/O looper for asynchronous I/O on the socket. */
Looper* looper;
/* I/O descriptor for asynchronous I/O on the socket. */
LoopIo io[1];
/* Timer to use for reconnection attempts. */
LoopTimer reconnect_timer[1];
/* Head of the list of the active readers. */
AsyncSocketIO* readers_head;
/* Tail of the list of the active readers. */
AsyncSocketIO* readers_tail;
/* Head of the list of the active writers. */
AsyncSocketIO* writers_head;
/* Tail of the list of the active writers. */
AsyncSocketIO* writers_tail;
/* Socket's file descriptor. */
int fd;
/* Timeout to use for reconnection attempts. */
int reconnect_to;
/* Number of outstanding references to the socket. */
int ref_count;
/* Flags whether (1) or not (0) socket owns the looper. */
int owns_looper;
};
static const char*
_async_socket_string(AsyncSocket* as)
{
return sock_address_to_string(&as->address);
}
static Looper*
_async_socket_get_looper(AsyncSocket* as)
{
return as->looper;
}
/* Pulls first reader out of the list.
* Param:
* as - Initialized AsyncSocket instance.
* Return:
* First I/O pulled out of the list, or NULL if there are no I/O in the list.
* Note that the caller is responsible for releasing the I/O object returned
* from this routine.
*/
static AsyncSocketIO*
_async_socket_pull_first_io(AsyncSocket* as,
AsyncSocketIO** list_head,
AsyncSocketIO** list_tail)
{
AsyncSocketIO* const ret = *list_head;
if (ret != NULL) {
*list_head = ret->next;
ret->next = NULL;
if (*list_head == NULL) {
*list_tail = NULL;
}
}
return ret;
}
/* Pulls first reader out of the list.
* Param:
* as - Initialized AsyncSocket instance.
* Return:
* First reader pulled out of the list, or NULL if there are no readers in the
* list.
* Note that the caller is responsible for releasing the I/O object returned
* from this routine.
*/
static AsyncSocketIO*
_async_socket_pull_first_reader(AsyncSocket* as)
{
return _async_socket_pull_first_io(as, &as->readers_head, &as->readers_tail);
}
/* Pulls first writer out of the list.
* Param:
* as - Initialized AsyncSocket instance.
* Return:
* First writer pulled out of the list, or NULL if there are no writers in the
* list.
* Note that the caller is responsible for releasing the I/O object returned
* from this routine.
*/
static AsyncSocketIO*
_async_socket_pull_first_writer(AsyncSocket* as)
{
return _async_socket_pull_first_io(as, &as->writers_head, &as->writers_tail);
}
/* Removes an I/O descriptor from a list of active I/O.
* Param:
* as - Initialized AsyncSocket instance.
* list_head, list_tail - Pointers to the list head and tail.
* io - I/O to remove.
* Return:
* Boolean: 1 if I/O has been removed, or 0 if I/O has not been found in the list.
*/
static int
_async_socket_remove_io(AsyncSocket* as,
AsyncSocketIO** list_head,
AsyncSocketIO** list_tail,
AsyncSocketIO* io)
{
AsyncSocketIO* prev = NULL;
while (*list_head != NULL && io != *list_head) {
prev = *list_head;
list_head = &((*list_head)->next);
}
if (*list_head == NULL) {
D("%s: I/O %p is not found in the list for socket '%s'",
__FUNCTION__, io, _async_socket_string(as));
return 0;
}
*list_head = io->next;
if (prev != NULL) {
prev->next = io->next;
}
if (*list_tail == io) {
*list_tail = prev;
}
/* Release I/O adjusting reference added when I/O has been saved in the list. */
async_socket_io_release(io);
return 1;
}
/* Advances to the next I/O in the list.
* Param:
* as - Initialized AsyncSocket instance.
* list_head, list_tail - Pointers to the list head and tail.
*/
static void
_async_socket_advance_io(AsyncSocket* as,
AsyncSocketIO** list_head,
AsyncSocketIO** list_tail)
{
AsyncSocketIO* first_io = *list_head;
if (first_io != NULL) {
*list_head = first_io->next;
first_io->next = NULL;
}
if (*list_head == NULL) {
*list_tail = NULL;
}
if (first_io != NULL) {
/* Release I/O removed from the head of the list. */
async_socket_io_release(first_io);
}
}
/* Advances to the next reader in the list.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_async_socket_advance_reader(AsyncSocket* as)
{
_async_socket_advance_io(as, &as->readers_head, &as->readers_tail);
}
/* Advances to the next writer in the list.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_async_socket_advance_writer(AsyncSocket* as)
{
_async_socket_advance_io(as, &as->writers_head, &as->writers_tail);
}
/* Completes an I/O.
* Param:
* as - Initialized AsyncSocket instance.
* asio - I/O to complete.
* Return:
* One of AsyncIOAction values.
*/
static AsyncIOAction
_async_socket_complete_io(AsyncSocket* as, AsyncSocketIO* asio)
{
T("ASocket %s: %s I/O %p is completed.",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE", asio);
/* Stop the timer. */
async_socket_io_cancel_time_out(asio);
return asio->on_io(asio->io_opaque, asio, ASIO_STATE_SUCCEEDED);
}
/* Timeouts an I/O.
* Param:
* as - Initialized AsyncSocket instance.
* asio - An I/O that has timed out.
* Return:
* One of AsyncIOAction values.
*/
static AsyncIOAction
_async_socket_io_timed_out(AsyncSocket* as, AsyncSocketIO* asio)
{
T("ASocket %s: %s I/O %p with deadline %lld has timed out at %lld",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE", asio,
asio->deadline, async_socket_deadline(as, 0));
/* Report to the client. */
const AsyncIOAction action = asio->on_io(asio->io_opaque, asio,
ASIO_STATE_TIMED_OUT);
/* Remove the I/O from a list of active I/O for actions other than retry. */
if (action != ASIO_ACTION_RETRY) {
if (asio->is_io_read) {
_async_socket_remove_io(as, &as->readers_head, &as->readers_tail, asio);
} else {
_async_socket_remove_io(as, &as->writers_head, &as->writers_tail, asio);
}
}
return action;
}
/* Cancels an I/O.
* Param:
* as - Initialized AsyncSocket instance.
* asio - An I/O to cancel.
* Return:
* One of AsyncIOAction values.
*/
static AsyncIOAction
_async_socket_cancel_io(AsyncSocket* as, AsyncSocketIO* asio)
{
T("ASocket %s: %s I/O %p is cancelled.",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE", asio);
/* Stop the timer. */
async_socket_io_cancel_time_out(asio);
return asio->on_io(asio->io_opaque, asio, ASIO_STATE_CANCELLED);
}
/* Reports an I/O failure.
* Param:
* as - Initialized AsyncSocket instance.
* asio - An I/O that has failed. Can be NULL for general failures.
* failure - Failure (errno) that has occurred.
* Return:
* One of AsyncIOAction values.
*/
static AsyncIOAction
_async_socket_io_failure(AsyncSocket* as, AsyncSocketIO* asio, int failure)
{
T("ASocket %s: %s I/O %p has failed: %d -> %s",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE", asio,
failure, strerror(failure));
/* Stop the timer. */
async_socket_io_cancel_time_out(asio);
errno = failure;
return asio->on_io(asio->io_opaque, asio, ASIO_STATE_FAILED);
}
/* Cancels all the active socket readers.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_async_socket_cancel_readers(AsyncSocket* as)
{
while (as->readers_head != NULL) {
AsyncSocketIO* const to_cancel = _async_socket_pull_first_reader(as);
/* We ignore action returned from the cancellation callback, since we're
* in a disconnected state here. */
_async_socket_cancel_io(as, to_cancel);
async_socket_io_release(to_cancel);
}
}
/* Cancels all the active socket writers.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_async_socket_cancel_writers(AsyncSocket* as)
{
while (as->writers_head != NULL) {
AsyncSocketIO* const to_cancel = _async_socket_pull_first_writer(as);
/* We ignore action returned from the cancellation callback, since we're
* in a disconnected state here. */
_async_socket_cancel_io(as, to_cancel);
async_socket_io_release(to_cancel);
}
}
/* Cancels all the I/O on the socket. */
static void
_async_socket_cancel_all_io(AsyncSocket* as)
{
/* Stop the reconnection timer. */
loopTimer_stop(as->reconnect_timer);
/* Stop read / write on the socket. */
loopIo_dontWantWrite(as->io);
loopIo_dontWantRead(as->io);
/* Cancel active readers and writers. */
_async_socket_cancel_readers(as);
_async_socket_cancel_writers(as);
}
/* Closes socket handle used by the async socket.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_async_socket_close_socket(AsyncSocket* as)
{
if (as->fd >= 0) {
T("ASocket %s: Socket handle %d is closed.",
_async_socket_string(as), as->fd);
loopIo_done(as->io);
socket_close(as->fd);
as->fd = -1;
}
}
/* Destroys AsyncSocket instance.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_async_socket_free(AsyncSocket* as)
{
if (as != NULL) {
T("ASocket %s: Socket descriptor is destroyed.", _async_socket_string(as));
/* Close socket. */
_async_socket_close_socket(as);
/* Free allocated resources. */
if (as->looper != NULL) {
loopTimer_done(as->reconnect_timer);
if (as->owns_looper) {
looper_free(as->looper);
}
}
sock_address_done(&as->address);
AFREE(as);
}
}
/* Starts reconnection attempts after connection has been lost.
* Param:
* as - Initialized AsyncSocket instance.
* to - Milliseconds to wait before reconnection attempt.
*/
static void
_async_socket_reconnect(AsyncSocket* as, int to)
{
T("ASocket %s: reconnecting in %dms...", _async_socket_string(as), to);
/* Make sure that no I/O is active, and socket is closed before we
* reconnect. */
_async_socket_cancel_all_io(as);
/* Set the timer for reconnection attempt. */
loopTimer_startRelative(as->reconnect_timer, to);
}
/********************************************************************************
* Asynchronous Socket callbacks
*******************************************************************************/
/* A callback that is invoked when socket gets disconnected.
* Param:
* as - Initialized AsyncSocket instance.
*/
static void
_on_async_socket_disconnected(AsyncSocket* as)
{
/* Save error to restore it for the client's callback. */
const int save_errno = errno;
AsyncIOAction action = ASIO_ACTION_ABORT;
D("ASocket %s: Disconnected.", _async_socket_string(as));
/* Cancel all the I/O on this socket. */
_async_socket_cancel_all_io(as);
/* Close the socket. */
_async_socket_close_socket(as);
/* Restore errno, and invoke client's callback. */
errno = save_errno;
action = as->on_connection(as->client_opaque, as, ASIO_STATE_FAILED);
if (action == ASIO_ACTION_RETRY) {
/* Client requested reconnection. */
_async_socket_reconnect(as, as->reconnect_to);
}
}
/* A callback that is invoked on socket's I/O failure.
* Param:
* as - Initialized AsyncSocket instance.
* asio - Descriptor for the failed I/O. Can be NULL for general failures.
*/
static AsyncIOAction
_on_async_socket_failure(AsyncSocket* as, AsyncSocketIO* asio)
{
D("ASocket %s: %s I/O failure: %d -> %s",
_async_socket_string(as), asio->is_io_read ? "READ" : "WRITE",
errno, strerror(errno));
/* Report the failure. */
return _async_socket_io_failure(as, asio, errno);
}
/* A callback that is invoked when there is data available to read.
* Param:
* as - Initialized AsyncSocket instance.
* Return:
* 0 on success, or -1 on failure. Failure returned from this routine will
* skip writes (if awailable) behind this read.
*/
static int
_on_async_socket_recv(AsyncSocket* as)
{
AsyncIOAction action;
/* Get current reader. */
AsyncSocketIO* const asr = as->readers_head;
if (asr == NULL) {
D("ASocket %s: No reader is available.", _async_socket_string(as));
loopIo_dontWantRead(as->io);
return 0;
}
/* Reference the reader while we're working with it in this callback. */
async_socket_io_reference(asr);
/* Bump I/O state, and inform the client that I/O is in progress. */
if (asr->state == ASIO_STATE_QUEUED) {
asr->state = ASIO_STATE_STARTED;
} else {
asr->state = ASIO_STATE_CONTINUES;
}
action = asr->on_io(asr->io_opaque, asr, asr->state);
if (action == ASIO_ACTION_ABORT) {
D("ASocket %s: Read is aborted by the client.", _async_socket_string(as));
/* Move on to the next reader. */
_async_socket_advance_reader(as);
/* Lets see if there are still active readers, and enable, or disable
* read I/O callback accordingly. */
if (as->readers_head != NULL) {
loopIo_wantRead(as->io);
} else {
loopIo_dontWantRead(as->io);
}
async_socket_io_release(asr);
return 0;
}
/* Read next chunk of data. */
int res = socket_recv(as->fd, asr->buffer + asr->transferred,
asr->to_transfer - asr->transferred);
while (res < 0 && errno == EINTR) {
res = socket_recv(as->fd, asr->buffer + asr->transferred,
asr->to_transfer - asr->transferred);
}
if (res == 0) {
/* Socket has been disconnected. */
errno = ECONNRESET;
_on_async_socket_disconnected(as);
async_socket_io_release(asr);
return -1;
}
if (res < 0) {
if (errno == EWOULDBLOCK || errno == EAGAIN) {
/* Yield to writes behind this read. */
loopIo_wantRead(as->io);
async_socket_io_release(asr);
return 0;
}
/* An I/O error. */
action = _on_async_socket_failure(as, asr);
if (action != ASIO_ACTION_RETRY) {
D("ASocket %s: Read is aborted on failure.", _async_socket_string(as));
/* Move on to the next reader. */
_async_socket_advance_reader(as);
/* Lets see if there are still active readers, and enable, or disable
* read I/O callback accordingly. */
if (as->readers_head != NULL) {
loopIo_wantRead(as->io);
} else {
loopIo_dontWantRead(as->io);
}
}
async_socket_io_release(asr);
return -1;
}
/* Update the reader's descriptor. */
asr->transferred += res;
if (asr->transferred == asr->to_transfer) {
/* This read is completed. Move on to the next reader. */
_async_socket_advance_reader(as);
/* Notify reader completion. */
_async_socket_complete_io(as, asr);
}
/* Lets see if there are still active readers, and enable, or disable read
* I/O callback accordingly. */
if (as->readers_head != NULL) {
loopIo_wantRead(as->io);
} else {
loopIo_dontWantRead(as->io);
}
async_socket_io_release(asr);
return 0;
}
/* A callback that is invoked when there is data available to write.
* Param:
* as - Initialized AsyncSocket instance.
* Return:
* 0 on success, or -1 on failure. Failure returned from this routine will
* skip reads (if awailable) behind this write.
*/
static int
_on_async_socket_send(AsyncSocket* as)
{
AsyncIOAction action;
/* Get current writer. */
AsyncSocketIO* const asw = as->writers_head;
if (asw == NULL) {
D("ASocket %s: No writer is available.", _async_socket_string(as));
loopIo_dontWantWrite(as->io);
return 0;
}
/* Reference the writer while we're working with it in this callback. */
async_socket_io_reference(asw);
/* Bump I/O state, and inform the client that I/O is in progress. */
if (asw->state == ASIO_STATE_QUEUED) {
asw->state = ASIO_STATE_STARTED;
} else {
asw->state = ASIO_STATE_CONTINUES;
}
action = asw->on_io(asw->io_opaque, asw, asw->state);
if (action == ASIO_ACTION_ABORT) {
D("ASocket %s: Write is aborted by the client.", _async_socket_string(as));
/* Move on to the next writer. */
_async_socket_advance_writer(as);
/* Lets see if there are still active writers, and enable, or disable
* write I/O callback accordingly. */
if (as->writers_head != NULL) {
loopIo_wantWrite(as->io);
} else {
loopIo_dontWantWrite(as->io);
}
async_socket_io_release(asw);
return 0;
}
/* Write next chunk of data. */
int res = socket_send(as->fd, asw->buffer + asw->transferred,
asw->to_transfer - asw->transferred);
while (res < 0 && errno == EINTR) {
res = socket_send(as->fd, asw->buffer + asw->transferred,
asw->to_transfer - asw->transferred);
}
if (res == 0) {
/* Socket has been disconnected. */
errno = ECONNRESET;
_on_async_socket_disconnected(as);
async_socket_io_release(asw);
return -1;
}
if (res < 0) {
if (errno == EWOULDBLOCK || errno == EAGAIN) {
/* Yield to reads behind this write. */
loopIo_wantWrite(as->io);
async_socket_io_release(asw);
return 0;
}
/* An I/O error. */
action = _on_async_socket_failure(as, asw);
if (action != ASIO_ACTION_RETRY) {
D("ASocket %s: Write is aborted on failure.", _async_socket_string(as));
/* Move on to the next writer. */
_async_socket_advance_writer(as);
/* Lets see if there are still active writers, and enable, or disable
* write I/O callback accordingly. */
if (as->writers_head != NULL) {
loopIo_wantWrite(as->io);
} else {
loopIo_dontWantWrite(as->io);
}
}
async_socket_io_release(asw);
return -1;
}
/* Update the writer descriptor. */
asw->transferred += res;
if (asw->transferred == asw->to_transfer) {
/* This write is completed. Move on to the next writer. */
_async_socket_advance_writer(as);
/* Notify writer completion. */
_async_socket_complete_io(as, asw);
}
/* Lets see if there are still active writers, and enable, or disable write
* I/O callback accordingly. */
if (as->writers_head != NULL) {
loopIo_wantWrite(as->io);
} else {
loopIo_dontWantWrite(as->io);
}
async_socket_io_release(asw);
return 0;
}
/* A callback that is invoked when an I/O is available on socket.
* Param:
* as - Initialized AsyncSocket instance.
* fd - Socket's file descriptor.
* events - LOOP_IO_READ | LOOP_IO_WRITE bitmask.
*/
static void
_on_async_socket_io(void* opaque, int fd, unsigned events)
{
AsyncSocket* const as = (AsyncSocket*)opaque;
/* Reference the socket while we're working with it in this callback. */
async_socket_reference(as);
if ((events & LOOP_IO_READ) != 0) {
if (_on_async_socket_recv(as) != 0) {
async_socket_release(as);
return;
}
}
if ((events & LOOP_IO_WRITE) != 0) {
if (_on_async_socket_send(as) != 0) {
async_socket_release(as);
return;
}
}
async_socket_release(as);
}
/* A callback that is invoked by asynchronous socket connector on connection
* events.
* Param:
* opaque - Initialized AsyncSocket instance.
* connector - Connector that is used to connect this socket.
* event - Connection event.
* Return:
* One of AsyncIOAction values.
*/
static AsyncIOAction
_on_connector_events(void* opaque,
AsyncSocketConnector* connector,
AsyncIOState event)
{
AsyncIOAction action;
AsyncSocket* const as = (AsyncSocket*)opaque;
/* Reference the socket while we're working with it in this callback. */
async_socket_reference(as);
if (event == ASIO_STATE_SUCCEEDED) {
/* Accept the connection. */
as->fd = async_socket_connector_pull_fd(connector);
loopIo_init(as->io, as->looper, as->fd, _on_async_socket_io, as);
}
/* Invoke client's callback. */
action = as->on_connection(as->client_opaque, as, event);
if (event == ASIO_STATE_SUCCEEDED && action != ASIO_ACTION_DONE) {
/* For whatever reason the client didn't want to keep this connection.
* Close it. */
D("ASocket %s: Connection is discarded by the client.",
_async_socket_string(as));
_async_socket_close_socket(as);
}
if (action != ASIO_ACTION_RETRY) {
async_socket_connector_release(connector);
}
async_socket_release(as);
return action;
}
/* Timer callback invoked to reconnect the lost connection.
* Param:
* as - Initialized AsyncSocket instance.
*/
void
_on_async_socket_reconnect(void* opaque)
{
AsyncSocket* as = (AsyncSocket*)opaque;
/* Reference the socket while we're working with it in this callback. */
async_socket_reference(as);
async_socket_connect(as, as->reconnect_to);
async_socket_release(as);
}
/********************************************************************************
* Android Device Socket public API
*******************************************************************************/
AsyncSocket*
async_socket_new(int port,
int reconnect_to,
on_as_connection_cb client_cb,
void* client_opaque,
Looper* looper)
{
AsyncSocket* as;
if (client_cb == NULL) {
E("Invalid client_cb parameter");
return NULL;
}
ANEW0(as);
as->fd = -1;
as->client_opaque = client_opaque;
as->on_connection = client_cb;
as->readers_head = as->readers_tail = NULL;
as->reconnect_to = reconnect_to;
as->ref_count = 1;
sock_address_init_inet(&as->address, SOCK_ADDRESS_INET_LOOPBACK, port);
if (looper == NULL) {
as->looper = looper_newCore();
if (as->looper == NULL) {
E("Unable to create I/O looper for async socket '%s'",
_async_socket_string(as));
client_cb(client_opaque, as, ASIO_STATE_FAILED);
_async_socket_free(as);
return NULL;
}
as->owns_looper = 1;
} else {
as->looper = looper;
as->owns_looper = 0;
}
loopTimer_init(as->reconnect_timer, as->looper, _on_async_socket_reconnect, as);
T("ASocket %s: Descriptor is created.", _async_socket_string(as));
return as;
}
int
async_socket_reference(AsyncSocket* as)
{
assert(as->ref_count > 0);
as->ref_count++;
return as->ref_count;
}
int
async_socket_release(AsyncSocket* as)
{
assert(as->ref_count > 0);
as->ref_count--;
if (as->ref_count == 0) {
/* Last reference has been dropped. Destroy this object. */
_async_socket_cancel_all_io(as);
_async_socket_free(as);
return 0;
}
return as->ref_count;
}
void
async_socket_connect(AsyncSocket* as, int retry_to)
{
T("ASocket %s: Handling connection request for %dms...",
_async_socket_string(as), retry_to);
AsyncSocketConnector* const connector =
async_socket_connector_new(&as->address, retry_to, _on_connector_events,
as, as->looper);
if (connector != NULL) {
async_socket_connector_connect(connector);
} else {
as->on_connection(as->client_opaque, as, ASIO_STATE_FAILED);
}
}
void
async_socket_disconnect(AsyncSocket* as)
{
T("ASocket %s: Handling disconnection request...", _async_socket_string(as));
if (as != NULL) {
_async_socket_cancel_all_io(as);
_async_socket_close_socket(as);
}
}
void
async_socket_reconnect(AsyncSocket* as, int retry_to)
{
T("ASocket %s: Handling reconnection request for %dms...",
_async_socket_string(as), retry_to);
_async_socket_cancel_all_io(as);
_async_socket_close_socket(as);
_async_socket_reconnect(as, retry_to);
}
void
async_socket_read_abs(AsyncSocket* as,
void* buffer, uint32_t len,
on_as_io_cb reader_cb,
void* reader_opaque,
Duration deadline)
{
T("ASocket %s: Handling read for %d bytes with deadline %lld...",
_async_socket_string(as), len, deadline);
AsyncSocketIO* const asr =
_async_socket_reader_new(as, buffer, len, reader_cb, reader_opaque,
deadline);
if (async_socket_is_connected(as)) {
/* Add new reader to the list. Note that we use initial reference from I/O
* 'new' routine as "in the list" reference counter. */
if (as->readers_head == NULL) {
as->readers_head = as->readers_tail = asr;
} else {
as->readers_tail->next = asr;
as->readers_tail = asr;
}
loopIo_wantRead(as->io);
} else {
D("ASocket %s: Read on a disconnected socket.", _async_socket_string(as));
errno = ECONNRESET;
reader_cb(reader_opaque, asr, ASIO_STATE_FAILED);
async_socket_io_release(asr);
}
}
void
async_socket_read_rel(AsyncSocket* as,
void* buffer, uint32_t len,
on_as_io_cb reader_cb,
void* reader_opaque,
int to)
{
const Duration dl = (to >= 0) ? looper_now(_async_socket_get_looper(as)) + to :
DURATION_INFINITE;
async_socket_read_abs(as, buffer, len, reader_cb, reader_opaque, dl);
}
void
async_socket_write_abs(AsyncSocket* as,
const void* buffer, uint32_t len,
on_as_io_cb writer_cb,
void* writer_opaque,
Duration deadline)
{
T("ASocket %s: Handling write for %d bytes with deadline %lld...",
_async_socket_string(as), len, deadline);
AsyncSocketIO* const asw =
_async_socket_writer_new(as, buffer, len, writer_cb, writer_opaque,
deadline);
if (async_socket_is_connected(as)) {
/* Add new writer to the list. Note that we use initial reference from I/O
* 'new' routine as "in the list" reference counter. */
if (as->writers_head == NULL) {
as->writers_head = as->writers_tail = asw;
} else {
as->writers_tail->next = asw;
as->writers_tail = asw;
}
loopIo_wantWrite(as->io);
} else {
D("ASocket %s: Write on a disconnected socket.", _async_socket_string(as));
errno = ECONNRESET;
writer_cb(writer_opaque, asw, ASIO_STATE_FAILED);
async_socket_io_release(asw);
}
}
void
async_socket_write_rel(AsyncSocket* as,
const void* buffer, uint32_t len,
on_as_io_cb writer_cb,
void* writer_opaque,
int to)
{
const Duration dl = (to >= 0) ? looper_now(_async_socket_get_looper(as)) + to :
DURATION_INFINITE;
async_socket_write_abs(as, buffer, len, writer_cb, writer_opaque, dl);
}
void*
async_socket_get_client_opaque(const AsyncSocket* as)
{
return as->client_opaque;
}
Duration
async_socket_deadline(AsyncSocket* as, int rel)
{
return (rel >= 0) ? looper_now(_async_socket_get_looper(as)) + rel :
DURATION_INFINITE;
}
int
async_socket_get_port(const AsyncSocket* as)
{
return sock_address_get_port(&as->address);
}
int
async_socket_is_connected(const AsyncSocket* as)
{
return as->fd >= 0;
}