123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356 |
- /*
- * Copyright (C) 2007 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.
- */
- #define LOG_TAG "mq"
- #include <assert.h>
- #include <errno.h>
- #include <fcntl.h>
- #include <pthread.h>
- #include <stdlib.h>
- #include <string.h>
- #include <unistd.h>
- #include <sys/socket.h>
- #include <sys/types.h>
- #include <sys/un.h>
- #include <sys/uio.h>
- #include <cutils/array.h>
- #include <cutils/hashmap.h>
- #include <cutils/selector.h>
- #include "loghack.h"
- #include "buffer.h"
- /** Number of dead peers to remember. */
- #define PEER_HISTORY (16)
- typedef struct sockaddr SocketAddress;
- typedef struct sockaddr_un UnixAddress;
- /**
- * Process/user/group ID. We don't use ucred directly because it's only
- * available on Linux.
- */
- typedef struct {
- pid_t pid;
- uid_t uid;
- gid_t gid;
- } Credentials;
- /** Listens for bytes coming from remote peers. */
- typedef void BytesListener(Credentials credentials, char* bytes, size_t size);
- /** Listens for the deaths of remote peers. */
- typedef void DeathListener(pid_t pid);
- /** Types of packets. */
- typedef enum {
- /** Request for a connection to another peer. */
- CONNECTION_REQUEST,
- /** A connection to another peer. */
- CONNECTION,
- /** Reports a failed connection attempt. */
- CONNECTION_ERROR,
- /** A generic packet of bytes. */
- BYTES,
- } PacketType;
- typedef enum {
- /** Reading a packet header. */
- READING_HEADER,
- /** Waiting for a connection from the master. */
- ACCEPTING_CONNECTION,
- /** Reading bytes. */
- READING_BYTES,
- } InputState;
- /** A packet header. */
- // TODO: Use custom headers for master->peer, peer->master, peer->peer.
- typedef struct {
- PacketType type;
- union {
- /** Packet size. Used for BYTES. */
- size_t size;
- /** Credentials. Used for CONNECTION and CONNECTION_REQUEST. */
- Credentials credentials;
- };
- } Header;
- /** A packet which will be sent to a peer. */
- typedef struct OutgoingPacket OutgoingPacket;
- struct OutgoingPacket {
- /** Packet header. */
- Header header;
- union {
- /** Connection to peer. Used with CONNECTION. */
- int socket;
- /** Buffer of bytes. Used with BYTES. */
- Buffer* bytes;
- };
- /** Frees all resources associated with this packet. */
- void (*free)(OutgoingPacket* packet);
- /** Optional context. */
- void* context;
- /** Next packet in the queue. */
- OutgoingPacket* nextPacket;
- };
- /** Represents a remote peer. */
- typedef struct PeerProxy PeerProxy;
- /** Local peer state. You typically have one peer per process. */
- typedef struct {
- /** This peer's PID. */
- pid_t pid;
- /**
- * Map from pid to peer proxy. The peer has a peer proxy for each remote
- * peer it's connected to.
- *
- * Acquire mutex before use.
- */
- Hashmap* peerProxies;
- /** Manages I/O. */
- Selector* selector;
- /** Used to synchronize operations with the selector thread. */
- pthread_mutex_t mutex;
- /** Is this peer the master? */
- bool master;
- /** Peer proxy for the master. */
- PeerProxy* masterProxy;
- /** Listens for packets from remote peers. */
- BytesListener* onBytes;
- /** Listens for deaths of remote peers. */
- DeathListener* onDeath;
- /** Keeps track of recently dead peers. Requires mutex. */
- pid_t deadPeers[PEER_HISTORY];
- size_t deadPeerCursor;
- } Peer;
- struct PeerProxy {
- /** Credentials of the remote process. */
- Credentials credentials;
- /** Keeps track of data coming in from the remote peer. */
- InputState inputState;
- Buffer* inputBuffer;
- PeerProxy* connecting;
- /** File descriptor for this peer. */
- SelectableFd* fd;
- /**
- * Queue of packets to be written out to the remote peer.
- *
- * Requires mutex.
- */
- // TODO: Limit queue length.
- OutgoingPacket* currentPacket;
- OutgoingPacket* lastPacket;
- /** Used to write outgoing header. */
- Buffer outgoingHeader;
- /** True if this is the master's proxy. */
- bool master;
- /** Reference back to the local peer. */
- Peer* peer;
- /**
- * Used in master only. Maps this peer proxy to other peer proxies to
- * which the peer has been connected to. Maps pid to PeerProxy. Helps
- * keep track of which connections we've sent to whom.
- */
- Hashmap* connections;
- };
- /** Server socket path. */
- static const char* MASTER_PATH = "/master.peer";
- /** Credentials of the master peer. */
- static const Credentials MASTER_CREDENTIALS = {0, 0, 0};
- /** Creates a peer proxy and adds it to the peer proxy map. */
- static PeerProxy* peerProxyCreate(Peer* peer, Credentials credentials);
- /** Sets the non-blocking flag on a descriptor. */
- static void setNonBlocking(int fd) {
- int flags;
- if ((flags = fcntl(fd, F_GETFL, 0)) < 0) {
- LOG_ALWAYS_FATAL("fcntl() error: %s", strerror(errno));
- }
- if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
- LOG_ALWAYS_FATAL("fcntl() error: %s", strerror(errno));
- }
- }
- /** Closes a fd and logs a warning if the close fails. */
- static void closeWithWarning(int fd) {
- int result = close(fd);
- if (result == -1) {
- ALOGW("close() error: %s", strerror(errno));
- }
- }
- /** Hashes pid_t keys. */
- static int pidHash(void* key) {
- pid_t* pid = (pid_t*) key;
- return (int) (*pid);
- }
- /** Compares pid_t keys. */
- static bool pidEquals(void* keyA, void* keyB) {
- pid_t* a = (pid_t*) keyA;
- pid_t* b = (pid_t*) keyB;
- return *a == *b;
- }
- /** Gets the master address. Not thread safe. */
- static UnixAddress* getMasterAddress() {
- static UnixAddress masterAddress;
- static bool initialized = false;
- if (initialized == false) {
- masterAddress.sun_family = AF_LOCAL;
- strcpy(masterAddress.sun_path, MASTER_PATH);
- initialized = true;
- }
- return &masterAddress;
- }
- /** Gets exclusive access to the peer for this thread. */
- static void peerLock(Peer* peer) {
- pthread_mutex_lock(&peer->mutex);
- }
- /** Releases exclusive access to the peer. */
- static void peerUnlock(Peer* peer) {
- pthread_mutex_unlock(&peer->mutex);
- }
- /** Frees a simple, i.e. header-only, outgoing packet. */
- static void outgoingPacketFree(OutgoingPacket* packet) {
- ALOGD("Freeing outgoing packet.");
- free(packet);
- }
- /**
- * Prepare to read a new packet from the peer.
- */
- static void peerProxyExpectHeader(PeerProxy* peerProxy) {
- peerProxy->inputState = READING_HEADER;
- bufferPrepareForRead(peerProxy->inputBuffer, sizeof(Header));
- }
- /** Sets up the buffer for the outgoing header. */
- static void peerProxyPrepareOutgoingHeader(PeerProxy* peerProxy) {
- peerProxy->outgoingHeader.data
- = (char*) &(peerProxy->currentPacket->header);
- peerProxy->outgoingHeader.size = sizeof(Header);
- bufferPrepareForWrite(&peerProxy->outgoingHeader);
- }
- /** Adds a packet to the end of the queue. Callers must have the mutex. */
- static void peerProxyEnqueueOutgoingPacket(PeerProxy* peerProxy,
- OutgoingPacket* newPacket) {
- newPacket->nextPacket = NULL; // Just in case.
- if (peerProxy->currentPacket == NULL) {
- // The queue is empty.
- peerProxy->currentPacket = newPacket;
- peerProxy->lastPacket = newPacket;
- peerProxyPrepareOutgoingHeader(peerProxy);
- } else {
- peerProxy->lastPacket->nextPacket = newPacket;
- }
- }
- /** Takes the peer lock and enqueues the given packet. */
- static void peerProxyLockAndEnqueueOutgoingPacket(PeerProxy* peerProxy,
- OutgoingPacket* newPacket) {
- Peer* peer = peerProxy->peer;
- peerLock(peer);
- peerProxyEnqueueOutgoingPacket(peerProxy, newPacket);
- peerUnlock(peer);
- }
- /**
- * Frees current packet and moves to the next one. Returns true if there is
- * a next packet or false if the queue is empty.
- */
- static bool peerProxyNextPacket(PeerProxy* peerProxy) {
- Peer* peer = peerProxy->peer;
- peerLock(peer);
- OutgoingPacket* current = peerProxy->currentPacket;
- if (current == NULL) {
- // The queue is already empty.
- peerUnlock(peer);
- return false;
- }
- OutgoingPacket* next = current->nextPacket;
- peerProxy->currentPacket = next;
- current->nextPacket = NULL;
- current->free(current);
- if (next == NULL) {
- // The queue is empty.
- peerProxy->lastPacket = NULL;
- peerUnlock(peer);
- return false;
- } else {
- peerUnlock(peer);
- peerProxyPrepareOutgoingHeader(peerProxy);
- // TODO: Start writing next packet? It would reduce the number of
- // system calls, but we could also starve other peers.
- return true;
- }
- }
- /**
- * Checks whether a peer died recently.
- */
- static bool peerIsDead(Peer* peer, pid_t pid) {
- size_t i;
- for (i = 0; i < PEER_HISTORY; i++) {
- pid_t deadPeer = peer->deadPeers[i];
- if (deadPeer == 0) {
- return false;
- }
- if (deadPeer == pid) {
- return true;
- }
- }
- return false;
- }
- /**
- * Cleans up connection information.
- */
- static bool peerProxyRemoveConnection(void* key, void* value, void* context) {
- PeerProxy* deadPeer = (PeerProxy*) context;
- PeerProxy* otherPeer = (PeerProxy*) value;
- hashmapRemove(otherPeer->connections, &(deadPeer->credentials.pid));
- return true;
- }
- /**
- * Called when the peer dies.
- */
- static void peerProxyKill(PeerProxy* peerProxy, bool errnoIsSet) {
- if (errnoIsSet) {
- ALOGI("Peer %d died. errno: %s", peerProxy->credentials.pid,
- strerror(errno));
- } else {
- ALOGI("Peer %d died.", peerProxy->credentials.pid);
- }
- // If we lost the master, we're up a creek. We can't let this happen.
- if (peerProxy->master) {
- LOG_ALWAYS_FATAL("Lost connection to master.");
- }
- Peer* localPeer = peerProxy->peer;
- pid_t pid = peerProxy->credentials.pid;
- peerLock(localPeer);
- // Remember for awhile that the peer died.
- localPeer->deadPeers[localPeer->deadPeerCursor]
- = peerProxy->credentials.pid;
- localPeer->deadPeerCursor++;
- if (localPeer->deadPeerCursor == PEER_HISTORY) {
- localPeer->deadPeerCursor = 0;
- }
- // Remove from peer map.
- hashmapRemove(localPeer->peerProxies, &pid);
- // External threads can no longer get to this peer proxy, so we don't
- // need the lock anymore.
- peerUnlock(localPeer);
- // Remove the fd from the selector.
- if (peerProxy->fd != NULL) {
- peerProxy->fd->remove = true;
- }
- // Clear outgoing packet queue.
- while (peerProxyNextPacket(peerProxy)) {}
- bufferFree(peerProxy->inputBuffer);
- // This only applies to the master.
- if (peerProxy->connections != NULL) {
- // We can't leave these other maps pointing to freed memory.
- hashmapForEach(peerProxy->connections, &peerProxyRemoveConnection,
- peerProxy);
- hashmapFree(peerProxy->connections);
- }
- // Invoke death listener.
- localPeer->onDeath(pid);
- // Free the peer proxy itself.
- free(peerProxy);
- }
- static void peerProxyHandleError(PeerProxy* peerProxy, char* functionName) {
- if (errno == EINTR) {
- // Log interruptions but otherwise ignore them.
- ALOGW("%s() interrupted.", functionName);
- } else if (errno == EAGAIN) {
- ALOGD("EWOULDBLOCK");
- // Ignore.
- } else {
- ALOGW("Error returned by %s().", functionName);
- peerProxyKill(peerProxy, true);
- }
- }
- /**
- * Buffers output sent to a peer. May be called multiple times until the entire
- * buffer is filled. Returns true when the buffer is empty.
- */
- static bool peerProxyWriteFromBuffer(PeerProxy* peerProxy, Buffer* outgoing) {
- ssize_t size = bufferWrite(outgoing, peerProxy->fd->fd);
- if (size < 0) {
- peerProxyHandleError(peerProxy, "write");
- return false;
- } else {
- return bufferWriteComplete(outgoing);
- }
- }
- /** Writes packet bytes to peer. */
- static void peerProxyWriteBytes(PeerProxy* peerProxy) {
- Buffer* buffer = peerProxy->currentPacket->bytes;
- if (peerProxyWriteFromBuffer(peerProxy, buffer)) {
- ALOGD("Bytes written.");
- peerProxyNextPacket(peerProxy);
- }
- }
- /** Sends a socket to the peer. */
- static void peerProxyWriteConnection(PeerProxy* peerProxy) {
- int socket = peerProxy->currentPacket->socket;
- // Why does sending and receiving fds have to be such a PITA?
- struct msghdr msg;
- struct iovec iov[1];
- union {
- struct cmsghdr cm;
- char control[CMSG_SPACE(sizeof(int))];
- } control_un;
- struct cmsghdr *cmptr;
- msg.msg_control = control_un.control;
- msg.msg_controllen = sizeof(control_un.control);
- cmptr = CMSG_FIRSTHDR(&msg);
- cmptr->cmsg_len = CMSG_LEN(sizeof(int));
- cmptr->cmsg_level = SOL_SOCKET;
- cmptr->cmsg_type = SCM_RIGHTS;
- // Store the socket in the message.
- *((int *) CMSG_DATA(cmptr)) = peerProxy->currentPacket->socket;
- msg.msg_name = NULL;
- msg.msg_namelen = 0;
- iov[0].iov_base = "";
- iov[0].iov_len = 1;
- msg.msg_iov = iov;
- msg.msg_iovlen = 1;
- ssize_t result = sendmsg(peerProxy->fd->fd, &msg, 0);
- if (result < 0) {
- peerProxyHandleError(peerProxy, "sendmsg");
- } else {
- // Success. Queue up the next packet.
- peerProxyNextPacket(peerProxy);
- }
- }
- /**
- * Writes some outgoing data.
- */
- static void peerProxyWrite(SelectableFd* fd) {
- // TODO: Try to write header and body with one system call.
- PeerProxy* peerProxy = (PeerProxy*) fd->data;
- OutgoingPacket* current = peerProxy->currentPacket;
- if (current == NULL) {
- // We have nothing left to write.
- return;
- }
- // Write the header.
- Buffer* outgoingHeader = &peerProxy->outgoingHeader;
- bool headerWritten = bufferWriteComplete(outgoingHeader);
- if (!headerWritten) {
- ALOGD("Writing header...");
- headerWritten = peerProxyWriteFromBuffer(peerProxy, outgoingHeader);
- if (headerWritten) {
- ALOGD("Header written.");
- }
- }
- // Write body.
- if (headerWritten) {
- PacketType type = current->header.type;
- switch (type) {
- case CONNECTION:
- peerProxyWriteConnection(peerProxy);
- break;
- case BYTES:
- peerProxyWriteBytes(peerProxy);
- break;
- case CONNECTION_REQUEST:
- case CONNECTION_ERROR:
- // These packets consist solely of a header.
- peerProxyNextPacket(peerProxy);
- break;
- default:
- LOG_ALWAYS_FATAL("Unknown packet type: %d", type);
- }
- }
- }
- /**
- * Sets up a peer proxy's fd before we try to select() it.
- */
- static void peerProxyBeforeSelect(SelectableFd* fd) {
- ALOGD("Before select...");
- PeerProxy* peerProxy = (PeerProxy*) fd->data;
- peerLock(peerProxy->peer);
- bool hasPackets = peerProxy->currentPacket != NULL;
- peerUnlock(peerProxy->peer);
- if (hasPackets) {
- ALOGD("Packets found. Setting onWritable().");
- fd->onWritable = &peerProxyWrite;
- } else {
- // We have nothing to write.
- fd->onWritable = NULL;
- }
- }
- /** Prepare to read bytes from the peer. */
- static void peerProxyExpectBytes(PeerProxy* peerProxy, Header* header) {
- ALOGD("Expecting %d bytes.", header->size);
- peerProxy->inputState = READING_BYTES;
- if (bufferPrepareForRead(peerProxy->inputBuffer, header->size) == -1) {
- ALOGW("Couldn't allocate memory for incoming data. Size: %u",
- (unsigned int) header->size);
- // TODO: Ignore the packet and log a warning?
- peerProxyKill(peerProxy, false);
- }
- }
- /**
- * Gets a peer proxy for the given ID. Creates a peer proxy if necessary.
- * Sends a connection request to the master if desired.
- *
- * Returns NULL if an error occurs. Sets errno to EHOSTDOWN if the peer died
- * or ENOMEM if memory couldn't be allocated.
- */
- static PeerProxy* peerProxyGetOrCreate(Peer* peer, pid_t pid,
- bool requestConnection) {
- if (pid == peer->pid) {
- errno = EINVAL;
- return NULL;
- }
- if (peerIsDead(peer, pid)) {
- errno = EHOSTDOWN;
- return NULL;
- }
- PeerProxy* peerProxy = hashmapGet(peer->peerProxies, &pid);
- if (peerProxy != NULL) {
- return peerProxy;
- }
- // If this is the master peer, we already know about all peers.
- if (peer->master) {
- errno = EHOSTDOWN;
- return NULL;
- }
- // Try to create a peer proxy.
- Credentials credentials;
- credentials.pid = pid;
- // Fake gid and uid until we have the real thing. The real creds are
- // filled in by masterProxyExpectConnection(). These fake creds will
- // never be exposed to the user.
- credentials.uid = 0;
- credentials.gid = 0;
- // Make sure we can allocate the connection request packet.
- OutgoingPacket* packet = NULL;
- if (requestConnection) {
- packet = calloc(1, sizeof(OutgoingPacket));
- if (packet == NULL) {
- errno = ENOMEM;
- return NULL;
- }
- packet->header.type = CONNECTION_REQUEST;
- packet->header.credentials = credentials;
- packet->free = &outgoingPacketFree;
- }
- peerProxy = peerProxyCreate(peer, credentials);
- if (peerProxy == NULL) {
- free(packet);
- errno = ENOMEM;
- return NULL;
- } else {
- // Send a connection request to the master.
- if (requestConnection) {
- PeerProxy* masterProxy = peer->masterProxy;
- peerProxyEnqueueOutgoingPacket(masterProxy, packet);
- }
- return peerProxy;
- }
- }
- /**
- * Switches the master peer proxy into a state where it's waiting for a
- * connection from the master.
- */
- static void masterProxyExpectConnection(PeerProxy* masterProxy,
- Header* header) {
- // TODO: Restructure things so we don't need this check.
- // Verify that this really is the master.
- if (!masterProxy->master) {
- ALOGW("Non-master process %d tried to send us a connection.",
- masterProxy->credentials.pid);
- // Kill off the evil peer.
- peerProxyKill(masterProxy, false);
- return;
- }
- masterProxy->inputState = ACCEPTING_CONNECTION;
- Peer* localPeer = masterProxy->peer;
- // Create a peer proxy so we have somewhere to stash the creds.
- // See if we already have a proxy set up.
- pid_t pid = header->credentials.pid;
- peerLock(localPeer);
- PeerProxy* peerProxy = peerProxyGetOrCreate(localPeer, pid, false);
- if (peerProxy == NULL) {
- ALOGW("Peer proxy creation failed: %s", strerror(errno));
- } else {
- // Fill in full credentials.
- peerProxy->credentials = header->credentials;
- }
- peerUnlock(localPeer);
- // Keep track of which peer proxy we're accepting a connection for.
- masterProxy->connecting = peerProxy;
- }
- /**
- * Reads input from a peer process.
- */
- static void peerProxyRead(SelectableFd* fd);
- /** Sets up fd callbacks. */
- static void peerProxySetFd(PeerProxy* peerProxy, SelectableFd* fd) {
- peerProxy->fd = fd;
- fd->data = peerProxy;
- fd->onReadable = &peerProxyRead;
- fd->beforeSelect = &peerProxyBeforeSelect;
- // Make the socket non-blocking.
- setNonBlocking(fd->fd);
- }
- /**
- * Accepts a connection sent by the master proxy.
- */
- static void masterProxyAcceptConnection(PeerProxy* masterProxy) {
- struct msghdr msg;
- struct iovec iov[1];
- ssize_t size;
- char ignored;
- int incomingFd;
- // TODO: Reuse code which writes the connection. Who the heck designed
- // this API anyway?
- union {
- struct cmsghdr cm;
- char control[CMSG_SPACE(sizeof(int))];
- } control_un;
- struct cmsghdr *cmptr;
- msg.msg_control = control_un.control;
- msg.msg_controllen = sizeof(control_un.control);
- msg.msg_name = NULL;
- msg.msg_namelen = 0;
- // We sent 1 byte of data so we can detect EOF.
- iov[0].iov_base = &ignored;
- iov[0].iov_len = 1;
- msg.msg_iov = iov;
- msg.msg_iovlen = 1;
- size = recvmsg(masterProxy->fd->fd, &msg, 0);
- if (size < 0) {
- if (errno == EINTR) {
- // Log interruptions but otherwise ignore them.
- ALOGW("recvmsg() interrupted.");
- return;
- } else if (errno == EAGAIN) {
- // Keep waiting for the connection.
- return;
- } else {
- LOG_ALWAYS_FATAL("Error reading connection from master: %s",
- strerror(errno));
- }
- } else if (size == 0) {
- // EOF.
- LOG_ALWAYS_FATAL("Received EOF from master.");
- }
- // Extract fd from message.
- if ((cmptr = CMSG_FIRSTHDR(&msg)) != NULL
- && cmptr->cmsg_len == CMSG_LEN(sizeof(int))) {
- if (cmptr->cmsg_level != SOL_SOCKET) {
- LOG_ALWAYS_FATAL("Expected SOL_SOCKET.");
- }
- if (cmptr->cmsg_type != SCM_RIGHTS) {
- LOG_ALWAYS_FATAL("Expected SCM_RIGHTS.");
- }
- incomingFd = *((int*) CMSG_DATA(cmptr));
- } else {
- LOG_ALWAYS_FATAL("Expected fd.");
- }
- // The peer proxy this connection is for.
- PeerProxy* peerProxy = masterProxy->connecting;
- if (peerProxy == NULL) {
- ALOGW("Received connection for unknown peer.");
- closeWithWarning(incomingFd);
- } else {
- Peer* peer = masterProxy->peer;
- SelectableFd* selectableFd = selectorAdd(peer->selector, incomingFd);
- if (selectableFd == NULL) {
- ALOGW("Error adding fd to selector for %d.",
- peerProxy->credentials.pid);
- closeWithWarning(incomingFd);
- peerProxyKill(peerProxy, false);
- }
- peerProxySetFd(peerProxy, selectableFd);
- }
- peerProxyExpectHeader(masterProxy);
- }
- /**
- * Frees an outgoing packet containing a connection.
- */
- static void outgoingPacketFreeSocket(OutgoingPacket* packet) {
- closeWithWarning(packet->socket);
- outgoingPacketFree(packet);
- }
- /**
- * Connects two known peers.
- */
- static void masterConnectPeers(PeerProxy* peerA, PeerProxy* peerB) {
- int sockets[2];
- int result = socketpair(AF_LOCAL, SOCK_STREAM, 0, sockets);
- if (result == -1) {
- ALOGW("socketpair() error: %s", strerror(errno));
- // TODO: Send CONNECTION_FAILED packets to peers.
- return;
- }
- OutgoingPacket* packetA = calloc(1, sizeof(OutgoingPacket));
- OutgoingPacket* packetB = calloc(1, sizeof(OutgoingPacket));
- if (packetA == NULL || packetB == NULL) {
- free(packetA);
- free(packetB);
- ALOGW("malloc() error. Failed to tell process %d that process %d is"
- " dead.", peerA->credentials.pid, peerB->credentials.pid);
- return;
- }
- packetA->header.type = CONNECTION;
- packetB->header.type = CONNECTION;
- packetA->header.credentials = peerB->credentials;
- packetB->header.credentials = peerA->credentials;
- packetA->socket = sockets[0];
- packetB->socket = sockets[1];
- packetA->free = &outgoingPacketFreeSocket;
- packetB->free = &outgoingPacketFreeSocket;
- peerLock(peerA->peer);
- peerProxyEnqueueOutgoingPacket(peerA, packetA);
- peerProxyEnqueueOutgoingPacket(peerB, packetB);
- peerUnlock(peerA->peer);
- }
- /**
- * Informs a peer that the peer they're trying to connect to couldn't be
- * found.
- */
- static void masterReportConnectionError(PeerProxy* peerProxy,
- Credentials credentials) {
- OutgoingPacket* packet = calloc(1, sizeof(OutgoingPacket));
- if (packet == NULL) {
- ALOGW("malloc() error. Failed to tell process %d that process %d is"
- " dead.", peerProxy->credentials.pid, credentials.pid);
- return;
- }
- packet->header.type = CONNECTION_ERROR;
- packet->header.credentials = credentials;
- packet->free = &outgoingPacketFree;
- peerProxyLockAndEnqueueOutgoingPacket(peerProxy, packet);
- }
- /**
- * Handles a request to be connected to another peer.
- */
- static void masterHandleConnectionRequest(PeerProxy* peerProxy,
- Header* header) {
- Peer* master = peerProxy->peer;
- pid_t targetPid = header->credentials.pid;
- if (!hashmapContainsKey(peerProxy->connections, &targetPid)) {
- // We haven't connected these peers yet.
- PeerProxy* targetPeer
- = (PeerProxy*) hashmapGet(master->peerProxies, &targetPid);
- if (targetPeer == NULL) {
- // Unknown process.
- masterReportConnectionError(peerProxy, header->credentials);
- } else {
- masterConnectPeers(peerProxy, targetPeer);
- }
- }
- // This packet is complete. Get ready for the next one.
- peerProxyExpectHeader(peerProxy);
- }
- /**
- * The master told us this peer is dead.
- */
- static void masterProxyHandleConnectionError(PeerProxy* masterProxy,
- Header* header) {
- Peer* peer = masterProxy->peer;
- // Look up the peer proxy.
- pid_t pid = header->credentials.pid;
- PeerProxy* peerProxy = NULL;
- peerLock(peer);
- peerProxy = hashmapGet(peer->peerProxies, &pid);
- peerUnlock(peer);
- if (peerProxy != NULL) {
- ALOGI("Couldn't connect to %d.", pid);
- peerProxyKill(peerProxy, false);
- } else {
- ALOGW("Peer proxy for %d not found. This shouldn't happen.", pid);
- }
- peerProxyExpectHeader(masterProxy);
- }
- /**
- * Handles a packet header.
- */
- static void peerProxyHandleHeader(PeerProxy* peerProxy, Header* header) {
- switch (header->type) {
- case CONNECTION_REQUEST:
- masterHandleConnectionRequest(peerProxy, header);
- break;
- case CONNECTION:
- masterProxyExpectConnection(peerProxy, header);
- break;
- case CONNECTION_ERROR:
- masterProxyHandleConnectionError(peerProxy, header);
- break;
- case BYTES:
- peerProxyExpectBytes(peerProxy, header);
- break;
- default:
- ALOGW("Invalid packet type from %d: %d", peerProxy->credentials.pid,
- header->type);
- peerProxyKill(peerProxy, false);
- }
- }
- /**
- * Buffers input sent by peer. May be called multiple times until the entire
- * buffer is filled. Returns true when the buffer is full.
- */
- static bool peerProxyBufferInput(PeerProxy* peerProxy) {
- Buffer* in = peerProxy->inputBuffer;
- ssize_t size = bufferRead(in, peerProxy->fd->fd);
- if (size < 0) {
- peerProxyHandleError(peerProxy, "read");
- return false;
- } else if (size == 0) {
- // EOF.
- ALOGI("EOF");
- peerProxyKill(peerProxy, false);
- return false;
- } else if (bufferReadComplete(in)) {
- // We're done!
- return true;
- } else {
- // Continue reading.
- return false;
- }
- }
- /**
- * Reads input from a peer process.
- */
- static void peerProxyRead(SelectableFd* fd) {
- ALOGD("Reading...");
- PeerProxy* peerProxy = (PeerProxy*) fd->data;
- int state = peerProxy->inputState;
- Buffer* in = peerProxy->inputBuffer;
- switch (state) {
- case READING_HEADER:
- if (peerProxyBufferInput(peerProxy)) {
- ALOGD("Header read.");
- // We've read the complete header.
- Header* header = (Header*) in->data;
- peerProxyHandleHeader(peerProxy, header);
- }
- break;
- case READING_BYTES:
- ALOGD("Reading bytes...");
- if (peerProxyBufferInput(peerProxy)) {
- ALOGD("Bytes read.");
- // We have the complete packet. Notify bytes listener.
- peerProxy->peer->onBytes(peerProxy->credentials,
- in->data, in->size);
- // Get ready for the next packet.
- peerProxyExpectHeader(peerProxy);
- }
- break;
- case ACCEPTING_CONNECTION:
- masterProxyAcceptConnection(peerProxy);
- break;
- default:
- LOG_ALWAYS_FATAL("Unknown state: %d", state);
- }
- }
- static PeerProxy* peerProxyCreate(Peer* peer, Credentials credentials) {
- PeerProxy* peerProxy = calloc(1, sizeof(PeerProxy));
- if (peerProxy == NULL) {
- return NULL;
- }
- peerProxy->inputBuffer = bufferCreate(sizeof(Header));
- if (peerProxy->inputBuffer == NULL) {
- free(peerProxy);
- return NULL;
- }
- peerProxy->peer = peer;
- peerProxy->credentials = credentials;
- // Initial state == expecting a header.
- peerProxyExpectHeader(peerProxy);
- // Add this proxy to the map. Make sure the key points to the stable memory
- // inside of the peer proxy itself.
- pid_t* pid = &(peerProxy->credentials.pid);
- hashmapPut(peer->peerProxies, pid, peerProxy);
- return peerProxy;
- }
- /** Accepts a connection to the master peer. */
- static void masterAcceptConnection(SelectableFd* listenerFd) {
- // Accept connection.
- int socket = accept(listenerFd->fd, NULL, NULL);
- if (socket == -1) {
- ALOGW("accept() error: %s", strerror(errno));
- return;
- }
- ALOGD("Accepted connection as fd %d.", socket);
- // Get credentials.
- Credentials credentials;
- struct ucred ucredentials;
- socklen_t credentialsSize = sizeof(struct ucred);
- int result = getsockopt(socket, SOL_SOCKET, SO_PEERCRED,
- &ucredentials, &credentialsSize);
- // We might want to verify credentialsSize.
- if (result == -1) {
- ALOGW("getsockopt() error: %s", strerror(errno));
- closeWithWarning(socket);
- return;
- }
- // Copy values into our own structure so we know we have the types right.
- credentials.pid = ucredentials.pid;
- credentials.uid = ucredentials.uid;
- credentials.gid = ucredentials.gid;
- ALOGI("Accepted connection from process %d.", credentials.pid);
- Peer* masterPeer = (Peer*) listenerFd->data;
- peerLock(masterPeer);
- // Make sure we don't already have a connection from that process.
- PeerProxy* peerProxy
- = hashmapGet(masterPeer->peerProxies, &credentials.pid);
- if (peerProxy != NULL) {
- peerUnlock(masterPeer);
- ALOGW("Alread connected to process %d.", credentials.pid);
- closeWithWarning(socket);
- return;
- }
- // Add connection to the selector.
- SelectableFd* socketFd = selectorAdd(masterPeer->selector, socket);
- if (socketFd == NULL) {
- peerUnlock(masterPeer);
- ALOGW("malloc() failed.");
- closeWithWarning(socket);
- return;
- }
- // Create a peer proxy.
- peerProxy = peerProxyCreate(masterPeer, credentials);
- peerUnlock(masterPeer);
- if (peerProxy == NULL) {
- ALOGW("malloc() failed.");
- socketFd->remove = true;
- closeWithWarning(socket);
- }
- peerProxy->connections = hashmapCreate(10, &pidHash, &pidEquals);
- peerProxySetFd(peerProxy, socketFd);
- }
- /**
- * Creates the local peer.
- */
- static Peer* peerCreate() {
- Peer* peer = calloc(1, sizeof(Peer));
- if (peer == NULL) {
- LOG_ALWAYS_FATAL("malloc() error.");
- }
- peer->peerProxies = hashmapCreate(10, &pidHash, &pidEquals);
- peer->selector = selectorCreate();
- pthread_mutexattr_t attributes;
- if (pthread_mutexattr_init(&attributes) != 0) {
- LOG_ALWAYS_FATAL("pthread_mutexattr_init() error.");
- }
- if (pthread_mutexattr_settype(&attributes, PTHREAD_MUTEX_RECURSIVE) != 0) {
- LOG_ALWAYS_FATAL("pthread_mutexattr_settype() error.");
- }
- if (pthread_mutex_init(&peer->mutex, &attributes) != 0) {
- LOG_ALWAYS_FATAL("pthread_mutex_init() error.");
- }
- peer->pid = getpid();
- return peer;
- }
- /** The local peer. */
- static Peer* localPeer;
- /** Frees a packet of bytes. */
- static void outgoingPacketFreeBytes(OutgoingPacket* packet) {
- ALOGD("Freeing outgoing packet.");
- bufferFree(packet->bytes);
- free(packet);
- }
- /**
- * Sends a packet of bytes to a remote peer. Returns 0 on success.
- *
- * Returns -1 if an error occurs. Sets errno to ENOMEM if memory couldn't be
- * allocated. Sets errno to EHOSTDOWN if the peer died recently. Sets errno
- * to EINVAL if pid is the same as the local pid.
- */
- int peerSendBytes(pid_t pid, const char* bytes, size_t size) {
- Peer* peer = localPeer;
- assert(peer != NULL);
- OutgoingPacket* packet = calloc(1, sizeof(OutgoingPacket));
- if (packet == NULL) {
- errno = ENOMEM;
- return -1;
- }
- Buffer* copy = bufferCreate(size);
- if (copy == NULL) {
- free(packet);
- errno = ENOMEM;
- return -1;
- }
- // Copy data.
- memcpy(copy->data, bytes, size);
- copy->size = size;
- packet->bytes = copy;
- packet->header.type = BYTES;
- packet->header.size = size;
- packet->free = outgoingPacketFreeBytes;
- bufferPrepareForWrite(packet->bytes);
- peerLock(peer);
- PeerProxy* peerProxy = peerProxyGetOrCreate(peer, pid, true);
- if (peerProxy == NULL) {
- // The peer is already dead or we couldn't alloc memory. Either way,
- // errno is set.
- peerUnlock(peer);
- packet->free(packet);
- return -1;
- } else {
- peerProxyEnqueueOutgoingPacket(peerProxy, packet);
- peerUnlock(peer);
- selectorWakeUp(peer->selector);
- return 0;
- }
- }
- /** Keeps track of how to free shared bytes. */
- typedef struct {
- void (*free)(void* context);
- void* context;
- } SharedBytesFreer;
- /** Frees shared bytes. */
- static void outgoingPacketFreeSharedBytes(OutgoingPacket* packet) {
- SharedBytesFreer* sharedBytesFreer
- = (SharedBytesFreer*) packet->context;
- sharedBytesFreer->free(sharedBytesFreer->context);
- free(sharedBytesFreer);
- free(packet);
- }
- /**
- * Sends a packet of bytes to a remote peer without copying the bytes. Calls
- * free() with context after the bytes have been sent.
- *
- * Returns -1 if an error occurs. Sets errno to ENOMEM if memory couldn't be
- * allocated. Sets errno to EHOSTDOWN if the peer died recently. Sets errno
- * to EINVAL if pid is the same as the local pid.
- */
- int peerSendSharedBytes(pid_t pid, char* bytes, size_t size,
- void (*free)(void* context), void* context) {
- Peer* peer = localPeer;
- assert(peer != NULL);
- OutgoingPacket* packet = calloc(1, sizeof(OutgoingPacket));
- if (packet == NULL) {
- errno = ENOMEM;
- return -1;
- }
- Buffer* wrapper = bufferWrap(bytes, size, size);
- if (wrapper == NULL) {
- free(packet);
- errno = ENOMEM;
- return -1;
- }
- SharedBytesFreer* sharedBytesFreer = malloc(sizeof(SharedBytesFreer));
- if (sharedBytesFreer == NULL) {
- free(packet);
- free(wrapper);
- errno = ENOMEM;
- return -1;
- }
- sharedBytesFreer->free = free;
- sharedBytesFreer->context = context;
- packet->bytes = wrapper;
- packet->context = sharedBytesFreer;
- packet->header.type = BYTES;
- packet->header.size = size;
- packet->free = &outgoingPacketFreeSharedBytes;
- bufferPrepareForWrite(packet->bytes);
- peerLock(peer);
- PeerProxy* peerProxy = peerProxyGetOrCreate(peer, pid, true);
- if (peerProxy == NULL) {
- // The peer is already dead or we couldn't alloc memory. Either way,
- // errno is set.
- peerUnlock(peer);
- packet->free(packet);
- return -1;
- } else {
- peerProxyEnqueueOutgoingPacket(peerProxy, packet);
- peerUnlock(peer);
- selectorWakeUp(peer->selector);
- return 0;
- }
- }
- /**
- * Starts the master peer. The master peer differs from other peers in that
- * it is responsible for connecting the other peers. You can only have one
- * master peer.
- *
- * Goes into an I/O loop and does not return.
- */
- void masterPeerInitialize(BytesListener* bytesListener,
- DeathListener* deathListener) {
- // Create and bind socket.
- int listenerSocket = socket(AF_LOCAL, SOCK_STREAM, 0);
- if (listenerSocket == -1) {
- LOG_ALWAYS_FATAL("socket() error: %s", strerror(errno));
- }
- unlink(MASTER_PATH);
- int result = bind(listenerSocket, (SocketAddress*) getMasterAddress(),
- sizeof(UnixAddress));
- if (result == -1) {
- LOG_ALWAYS_FATAL("bind() error: %s", strerror(errno));
- }
- ALOGD("Listener socket: %d", listenerSocket);
- // Queue up to 16 connections.
- result = listen(listenerSocket, 16);
- if (result != 0) {
- LOG_ALWAYS_FATAL("listen() error: %s", strerror(errno));
- }
- // Make socket non-blocking.
- setNonBlocking(listenerSocket);
- // Create the peer for this process. Fail if we already have one.
- if (localPeer != NULL) {
- LOG_ALWAYS_FATAL("Peer is already initialized.");
- }
- localPeer = peerCreate();
- if (localPeer == NULL) {
- LOG_ALWAYS_FATAL("malloc() failed.");
- }
- localPeer->master = true;
- localPeer->onBytes = bytesListener;
- localPeer->onDeath = deathListener;
- // Make listener socket selectable.
- SelectableFd* listenerFd = selectorAdd(localPeer->selector, listenerSocket);
- if (listenerFd == NULL) {
- LOG_ALWAYS_FATAL("malloc() error.");
- }
- listenerFd->data = localPeer;
- listenerFd->onReadable = &masterAcceptConnection;
- }
- /**
- * Starts a local peer.
- *
- * Goes into an I/O loop and does not return.
- */
- void peerInitialize(BytesListener* bytesListener,
- DeathListener* deathListener) {
- // Connect to master peer.
- int masterSocket = socket(AF_LOCAL, SOCK_STREAM, 0);
- if (masterSocket == -1) {
- LOG_ALWAYS_FATAL("socket() error: %s", strerror(errno));
- }
- int result = connect(masterSocket, (SocketAddress*) getMasterAddress(),
- sizeof(UnixAddress));
- if (result != 0) {
- LOG_ALWAYS_FATAL("connect() error: %s", strerror(errno));
- }
- // Create the peer for this process. Fail if we already have one.
- if (localPeer != NULL) {
- LOG_ALWAYS_FATAL("Peer is already initialized.");
- }
- localPeer = peerCreate();
- if (localPeer == NULL) {
- LOG_ALWAYS_FATAL("malloc() failed.");
- }
- localPeer->onBytes = bytesListener;
- localPeer->onDeath = deathListener;
- // Make connection selectable.
- SelectableFd* masterFd = selectorAdd(localPeer->selector, masterSocket);
- if (masterFd == NULL) {
- LOG_ALWAYS_FATAL("malloc() error.");
- }
- // Create a peer proxy for the master peer.
- PeerProxy* masterProxy = peerProxyCreate(localPeer, MASTER_CREDENTIALS);
- if (masterProxy == NULL) {
- LOG_ALWAYS_FATAL("malloc() error.");
- }
- peerProxySetFd(masterProxy, masterFd);
- masterProxy->master = true;
- localPeer->masterProxy = masterProxy;
- }
- /** Starts the master peer I/O loop. Doesn't return. */
- void peerLoop() {
- assert(localPeer != NULL);
- // Start selector.
- selectorLoop(localPeer->selector);
- }
|