mq.c 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356
  1. /*
  2. * Copyright (C) 2007 The Android Open Source Project
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #define LOG_TAG "mq"
  17. #include <assert.h>
  18. #include <errno.h>
  19. #include <fcntl.h>
  20. #include <pthread.h>
  21. #include <stdlib.h>
  22. #include <string.h>
  23. #include <unistd.h>
  24. #include <sys/socket.h>
  25. #include <sys/types.h>
  26. #include <sys/un.h>
  27. #include <sys/uio.h>
  28. #include <cutils/array.h>
  29. #include <cutils/hashmap.h>
  30. #include <cutils/selector.h>
  31. #include "loghack.h"
  32. #include "buffer.h"
  33. /** Number of dead peers to remember. */
  34. #define PEER_HISTORY (16)
  35. typedef struct sockaddr SocketAddress;
  36. typedef struct sockaddr_un UnixAddress;
  37. /**
  38. * Process/user/group ID. We don't use ucred directly because it's only
  39. * available on Linux.
  40. */
  41. typedef struct {
  42. pid_t pid;
  43. uid_t uid;
  44. gid_t gid;
  45. } Credentials;
  46. /** Listens for bytes coming from remote peers. */
  47. typedef void BytesListener(Credentials credentials, char* bytes, size_t size);
  48. /** Listens for the deaths of remote peers. */
  49. typedef void DeathListener(pid_t pid);
  50. /** Types of packets. */
  51. typedef enum {
  52. /** Request for a connection to another peer. */
  53. CONNECTION_REQUEST,
  54. /** A connection to another peer. */
  55. CONNECTION,
  56. /** Reports a failed connection attempt. */
  57. CONNECTION_ERROR,
  58. /** A generic packet of bytes. */
  59. BYTES,
  60. } PacketType;
  61. typedef enum {
  62. /** Reading a packet header. */
  63. READING_HEADER,
  64. /** Waiting for a connection from the master. */
  65. ACCEPTING_CONNECTION,
  66. /** Reading bytes. */
  67. READING_BYTES,
  68. } InputState;
  69. /** A packet header. */
  70. // TODO: Use custom headers for master->peer, peer->master, peer->peer.
  71. typedef struct {
  72. PacketType type;
  73. union {
  74. /** Packet size. Used for BYTES. */
  75. size_t size;
  76. /** Credentials. Used for CONNECTION and CONNECTION_REQUEST. */
  77. Credentials credentials;
  78. };
  79. } Header;
  80. /** A packet which will be sent to a peer. */
  81. typedef struct OutgoingPacket OutgoingPacket;
  82. struct OutgoingPacket {
  83. /** Packet header. */
  84. Header header;
  85. union {
  86. /** Connection to peer. Used with CONNECTION. */
  87. int socket;
  88. /** Buffer of bytes. Used with BYTES. */
  89. Buffer* bytes;
  90. };
  91. /** Frees all resources associated with this packet. */
  92. void (*free)(OutgoingPacket* packet);
  93. /** Optional context. */
  94. void* context;
  95. /** Next packet in the queue. */
  96. OutgoingPacket* nextPacket;
  97. };
  98. /** Represents a remote peer. */
  99. typedef struct PeerProxy PeerProxy;
  100. /** Local peer state. You typically have one peer per process. */
  101. typedef struct {
  102. /** This peer's PID. */
  103. pid_t pid;
  104. /**
  105. * Map from pid to peer proxy. The peer has a peer proxy for each remote
  106. * peer it's connected to.
  107. *
  108. * Acquire mutex before use.
  109. */
  110. Hashmap* peerProxies;
  111. /** Manages I/O. */
  112. Selector* selector;
  113. /** Used to synchronize operations with the selector thread. */
  114. pthread_mutex_t mutex;
  115. /** Is this peer the master? */
  116. bool master;
  117. /** Peer proxy for the master. */
  118. PeerProxy* masterProxy;
  119. /** Listens for packets from remote peers. */
  120. BytesListener* onBytes;
  121. /** Listens for deaths of remote peers. */
  122. DeathListener* onDeath;
  123. /** Keeps track of recently dead peers. Requires mutex. */
  124. pid_t deadPeers[PEER_HISTORY];
  125. size_t deadPeerCursor;
  126. } Peer;
  127. struct PeerProxy {
  128. /** Credentials of the remote process. */
  129. Credentials credentials;
  130. /** Keeps track of data coming in from the remote peer. */
  131. InputState inputState;
  132. Buffer* inputBuffer;
  133. PeerProxy* connecting;
  134. /** File descriptor for this peer. */
  135. SelectableFd* fd;
  136. /**
  137. * Queue of packets to be written out to the remote peer.
  138. *
  139. * Requires mutex.
  140. */
  141. // TODO: Limit queue length.
  142. OutgoingPacket* currentPacket;
  143. OutgoingPacket* lastPacket;
  144. /** Used to write outgoing header. */
  145. Buffer outgoingHeader;
  146. /** True if this is the master's proxy. */
  147. bool master;
  148. /** Reference back to the local peer. */
  149. Peer* peer;
  150. /**
  151. * Used in master only. Maps this peer proxy to other peer proxies to
  152. * which the peer has been connected to. Maps pid to PeerProxy. Helps
  153. * keep track of which connections we've sent to whom.
  154. */
  155. Hashmap* connections;
  156. };
  157. /** Server socket path. */
  158. static const char* MASTER_PATH = "/master.peer";
  159. /** Credentials of the master peer. */
  160. static const Credentials MASTER_CREDENTIALS = {0, 0, 0};
  161. /** Creates a peer proxy and adds it to the peer proxy map. */
  162. static PeerProxy* peerProxyCreate(Peer* peer, Credentials credentials);
  163. /** Sets the non-blocking flag on a descriptor. */
  164. static void setNonBlocking(int fd) {
  165. int flags;
  166. if ((flags = fcntl(fd, F_GETFL, 0)) < 0) {
  167. LOG_ALWAYS_FATAL("fcntl() error: %s", strerror(errno));
  168. }
  169. if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
  170. LOG_ALWAYS_FATAL("fcntl() error: %s", strerror(errno));
  171. }
  172. }
  173. /** Closes a fd and logs a warning if the close fails. */
  174. static void closeWithWarning(int fd) {
  175. int result = close(fd);
  176. if (result == -1) {
  177. ALOGW("close() error: %s", strerror(errno));
  178. }
  179. }
  180. /** Hashes pid_t keys. */
  181. static int pidHash(void* key) {
  182. pid_t* pid = (pid_t*) key;
  183. return (int) (*pid);
  184. }
  185. /** Compares pid_t keys. */
  186. static bool pidEquals(void* keyA, void* keyB) {
  187. pid_t* a = (pid_t*) keyA;
  188. pid_t* b = (pid_t*) keyB;
  189. return *a == *b;
  190. }
  191. /** Gets the master address. Not thread safe. */
  192. static UnixAddress* getMasterAddress() {
  193. static UnixAddress masterAddress;
  194. static bool initialized = false;
  195. if (initialized == false) {
  196. masterAddress.sun_family = AF_LOCAL;
  197. strcpy(masterAddress.sun_path, MASTER_PATH);
  198. initialized = true;
  199. }
  200. return &masterAddress;
  201. }
  202. /** Gets exclusive access to the peer for this thread. */
  203. static void peerLock(Peer* peer) {
  204. pthread_mutex_lock(&peer->mutex);
  205. }
  206. /** Releases exclusive access to the peer. */
  207. static void peerUnlock(Peer* peer) {
  208. pthread_mutex_unlock(&peer->mutex);
  209. }
  210. /** Frees a simple, i.e. header-only, outgoing packet. */
  211. static void outgoingPacketFree(OutgoingPacket* packet) {
  212. ALOGD("Freeing outgoing packet.");
  213. free(packet);
  214. }
  215. /**
  216. * Prepare to read a new packet from the peer.
  217. */
  218. static void peerProxyExpectHeader(PeerProxy* peerProxy) {
  219. peerProxy->inputState = READING_HEADER;
  220. bufferPrepareForRead(peerProxy->inputBuffer, sizeof(Header));
  221. }
  222. /** Sets up the buffer for the outgoing header. */
  223. static void peerProxyPrepareOutgoingHeader(PeerProxy* peerProxy) {
  224. peerProxy->outgoingHeader.data
  225. = (char*) &(peerProxy->currentPacket->header);
  226. peerProxy->outgoingHeader.size = sizeof(Header);
  227. bufferPrepareForWrite(&peerProxy->outgoingHeader);
  228. }
  229. /** Adds a packet to the end of the queue. Callers must have the mutex. */
  230. static void peerProxyEnqueueOutgoingPacket(PeerProxy* peerProxy,
  231. OutgoingPacket* newPacket) {
  232. newPacket->nextPacket = NULL; // Just in case.
  233. if (peerProxy->currentPacket == NULL) {
  234. // The queue is empty.
  235. peerProxy->currentPacket = newPacket;
  236. peerProxy->lastPacket = newPacket;
  237. peerProxyPrepareOutgoingHeader(peerProxy);
  238. } else {
  239. peerProxy->lastPacket->nextPacket = newPacket;
  240. }
  241. }
  242. /** Takes the peer lock and enqueues the given packet. */
  243. static void peerProxyLockAndEnqueueOutgoingPacket(PeerProxy* peerProxy,
  244. OutgoingPacket* newPacket) {
  245. Peer* peer = peerProxy->peer;
  246. peerLock(peer);
  247. peerProxyEnqueueOutgoingPacket(peerProxy, newPacket);
  248. peerUnlock(peer);
  249. }
  250. /**
  251. * Frees current packet and moves to the next one. Returns true if there is
  252. * a next packet or false if the queue is empty.
  253. */
  254. static bool peerProxyNextPacket(PeerProxy* peerProxy) {
  255. Peer* peer = peerProxy->peer;
  256. peerLock(peer);
  257. OutgoingPacket* current = peerProxy->currentPacket;
  258. if (current == NULL) {
  259. // The queue is already empty.
  260. peerUnlock(peer);
  261. return false;
  262. }
  263. OutgoingPacket* next = current->nextPacket;
  264. peerProxy->currentPacket = next;
  265. current->nextPacket = NULL;
  266. current->free(current);
  267. if (next == NULL) {
  268. // The queue is empty.
  269. peerProxy->lastPacket = NULL;
  270. peerUnlock(peer);
  271. return false;
  272. } else {
  273. peerUnlock(peer);
  274. peerProxyPrepareOutgoingHeader(peerProxy);
  275. // TODO: Start writing next packet? It would reduce the number of
  276. // system calls, but we could also starve other peers.
  277. return true;
  278. }
  279. }
  280. /**
  281. * Checks whether a peer died recently.
  282. */
  283. static bool peerIsDead(Peer* peer, pid_t pid) {
  284. size_t i;
  285. for (i = 0; i < PEER_HISTORY; i++) {
  286. pid_t deadPeer = peer->deadPeers[i];
  287. if (deadPeer == 0) {
  288. return false;
  289. }
  290. if (deadPeer == pid) {
  291. return true;
  292. }
  293. }
  294. return false;
  295. }
  296. /**
  297. * Cleans up connection information.
  298. */
  299. static bool peerProxyRemoveConnection(void* key, void* value, void* context) {
  300. PeerProxy* deadPeer = (PeerProxy*) context;
  301. PeerProxy* otherPeer = (PeerProxy*) value;
  302. hashmapRemove(otherPeer->connections, &(deadPeer->credentials.pid));
  303. return true;
  304. }
  305. /**
  306. * Called when the peer dies.
  307. */
  308. static void peerProxyKill(PeerProxy* peerProxy, bool errnoIsSet) {
  309. if (errnoIsSet) {
  310. ALOGI("Peer %d died. errno: %s", peerProxy->credentials.pid,
  311. strerror(errno));
  312. } else {
  313. ALOGI("Peer %d died.", peerProxy->credentials.pid);
  314. }
  315. // If we lost the master, we're up a creek. We can't let this happen.
  316. if (peerProxy->master) {
  317. LOG_ALWAYS_FATAL("Lost connection to master.");
  318. }
  319. Peer* localPeer = peerProxy->peer;
  320. pid_t pid = peerProxy->credentials.pid;
  321. peerLock(localPeer);
  322. // Remember for awhile that the peer died.
  323. localPeer->deadPeers[localPeer->deadPeerCursor]
  324. = peerProxy->credentials.pid;
  325. localPeer->deadPeerCursor++;
  326. if (localPeer->deadPeerCursor == PEER_HISTORY) {
  327. localPeer->deadPeerCursor = 0;
  328. }
  329. // Remove from peer map.
  330. hashmapRemove(localPeer->peerProxies, &pid);
  331. // External threads can no longer get to this peer proxy, so we don't
  332. // need the lock anymore.
  333. peerUnlock(localPeer);
  334. // Remove the fd from the selector.
  335. if (peerProxy->fd != NULL) {
  336. peerProxy->fd->remove = true;
  337. }
  338. // Clear outgoing packet queue.
  339. while (peerProxyNextPacket(peerProxy)) {}
  340. bufferFree(peerProxy->inputBuffer);
  341. // This only applies to the master.
  342. if (peerProxy->connections != NULL) {
  343. // We can't leave these other maps pointing to freed memory.
  344. hashmapForEach(peerProxy->connections, &peerProxyRemoveConnection,
  345. peerProxy);
  346. hashmapFree(peerProxy->connections);
  347. }
  348. // Invoke death listener.
  349. localPeer->onDeath(pid);
  350. // Free the peer proxy itself.
  351. free(peerProxy);
  352. }
  353. static void peerProxyHandleError(PeerProxy* peerProxy, char* functionName) {
  354. if (errno == EINTR) {
  355. // Log interruptions but otherwise ignore them.
  356. ALOGW("%s() interrupted.", functionName);
  357. } else if (errno == EAGAIN) {
  358. ALOGD("EWOULDBLOCK");
  359. // Ignore.
  360. } else {
  361. ALOGW("Error returned by %s().", functionName);
  362. peerProxyKill(peerProxy, true);
  363. }
  364. }
  365. /**
  366. * Buffers output sent to a peer. May be called multiple times until the entire
  367. * buffer is filled. Returns true when the buffer is empty.
  368. */
  369. static bool peerProxyWriteFromBuffer(PeerProxy* peerProxy, Buffer* outgoing) {
  370. ssize_t size = bufferWrite(outgoing, peerProxy->fd->fd);
  371. if (size < 0) {
  372. peerProxyHandleError(peerProxy, "write");
  373. return false;
  374. } else {
  375. return bufferWriteComplete(outgoing);
  376. }
  377. }
  378. /** Writes packet bytes to peer. */
  379. static void peerProxyWriteBytes(PeerProxy* peerProxy) {
  380. Buffer* buffer = peerProxy->currentPacket->bytes;
  381. if (peerProxyWriteFromBuffer(peerProxy, buffer)) {
  382. ALOGD("Bytes written.");
  383. peerProxyNextPacket(peerProxy);
  384. }
  385. }
  386. /** Sends a socket to the peer. */
  387. static void peerProxyWriteConnection(PeerProxy* peerProxy) {
  388. int socket = peerProxy->currentPacket->socket;
  389. // Why does sending and receiving fds have to be such a PITA?
  390. struct msghdr msg;
  391. struct iovec iov[1];
  392. union {
  393. struct cmsghdr cm;
  394. char control[CMSG_SPACE(sizeof(int))];
  395. } control_un;
  396. struct cmsghdr *cmptr;
  397. msg.msg_control = control_un.control;
  398. msg.msg_controllen = sizeof(control_un.control);
  399. cmptr = CMSG_FIRSTHDR(&msg);
  400. cmptr->cmsg_len = CMSG_LEN(sizeof(int));
  401. cmptr->cmsg_level = SOL_SOCKET;
  402. cmptr->cmsg_type = SCM_RIGHTS;
  403. // Store the socket in the message.
  404. *((int *) CMSG_DATA(cmptr)) = peerProxy->currentPacket->socket;
  405. msg.msg_name = NULL;
  406. msg.msg_namelen = 0;
  407. iov[0].iov_base = "";
  408. iov[0].iov_len = 1;
  409. msg.msg_iov = iov;
  410. msg.msg_iovlen = 1;
  411. ssize_t result = sendmsg(peerProxy->fd->fd, &msg, 0);
  412. if (result < 0) {
  413. peerProxyHandleError(peerProxy, "sendmsg");
  414. } else {
  415. // Success. Queue up the next packet.
  416. peerProxyNextPacket(peerProxy);
  417. }
  418. }
  419. /**
  420. * Writes some outgoing data.
  421. */
  422. static void peerProxyWrite(SelectableFd* fd) {
  423. // TODO: Try to write header and body with one system call.
  424. PeerProxy* peerProxy = (PeerProxy*) fd->data;
  425. OutgoingPacket* current = peerProxy->currentPacket;
  426. if (current == NULL) {
  427. // We have nothing left to write.
  428. return;
  429. }
  430. // Write the header.
  431. Buffer* outgoingHeader = &peerProxy->outgoingHeader;
  432. bool headerWritten = bufferWriteComplete(outgoingHeader);
  433. if (!headerWritten) {
  434. ALOGD("Writing header...");
  435. headerWritten = peerProxyWriteFromBuffer(peerProxy, outgoingHeader);
  436. if (headerWritten) {
  437. ALOGD("Header written.");
  438. }
  439. }
  440. // Write body.
  441. if (headerWritten) {
  442. PacketType type = current->header.type;
  443. switch (type) {
  444. case CONNECTION:
  445. peerProxyWriteConnection(peerProxy);
  446. break;
  447. case BYTES:
  448. peerProxyWriteBytes(peerProxy);
  449. break;
  450. case CONNECTION_REQUEST:
  451. case CONNECTION_ERROR:
  452. // These packets consist solely of a header.
  453. peerProxyNextPacket(peerProxy);
  454. break;
  455. default:
  456. LOG_ALWAYS_FATAL("Unknown packet type: %d", type);
  457. }
  458. }
  459. }
  460. /**
  461. * Sets up a peer proxy's fd before we try to select() it.
  462. */
  463. static void peerProxyBeforeSelect(SelectableFd* fd) {
  464. ALOGD("Before select...");
  465. PeerProxy* peerProxy = (PeerProxy*) fd->data;
  466. peerLock(peerProxy->peer);
  467. bool hasPackets = peerProxy->currentPacket != NULL;
  468. peerUnlock(peerProxy->peer);
  469. if (hasPackets) {
  470. ALOGD("Packets found. Setting onWritable().");
  471. fd->onWritable = &peerProxyWrite;
  472. } else {
  473. // We have nothing to write.
  474. fd->onWritable = NULL;
  475. }
  476. }
  477. /** Prepare to read bytes from the peer. */
  478. static void peerProxyExpectBytes(PeerProxy* peerProxy, Header* header) {
  479. ALOGD("Expecting %d bytes.", header->size);
  480. peerProxy->inputState = READING_BYTES;
  481. if (bufferPrepareForRead(peerProxy->inputBuffer, header->size) == -1) {
  482. ALOGW("Couldn't allocate memory for incoming data. Size: %u",
  483. (unsigned int) header->size);
  484. // TODO: Ignore the packet and log a warning?
  485. peerProxyKill(peerProxy, false);
  486. }
  487. }
  488. /**
  489. * Gets a peer proxy for the given ID. Creates a peer proxy if necessary.
  490. * Sends a connection request to the master if desired.
  491. *
  492. * Returns NULL if an error occurs. Sets errno to EHOSTDOWN if the peer died
  493. * or ENOMEM if memory couldn't be allocated.
  494. */
  495. static PeerProxy* peerProxyGetOrCreate(Peer* peer, pid_t pid,
  496. bool requestConnection) {
  497. if (pid == peer->pid) {
  498. errno = EINVAL;
  499. return NULL;
  500. }
  501. if (peerIsDead(peer, pid)) {
  502. errno = EHOSTDOWN;
  503. return NULL;
  504. }
  505. PeerProxy* peerProxy = hashmapGet(peer->peerProxies, &pid);
  506. if (peerProxy != NULL) {
  507. return peerProxy;
  508. }
  509. // If this is the master peer, we already know about all peers.
  510. if (peer->master) {
  511. errno = EHOSTDOWN;
  512. return NULL;
  513. }
  514. // Try to create a peer proxy.
  515. Credentials credentials;
  516. credentials.pid = pid;
  517. // Fake gid and uid until we have the real thing. The real creds are
  518. // filled in by masterProxyExpectConnection(). These fake creds will
  519. // never be exposed to the user.
  520. credentials.uid = 0;
  521. credentials.gid = 0;
  522. // Make sure we can allocate the connection request packet.
  523. OutgoingPacket* packet = NULL;
  524. if (requestConnection) {
  525. packet = calloc(1, sizeof(OutgoingPacket));
  526. if (packet == NULL) {
  527. errno = ENOMEM;
  528. return NULL;
  529. }
  530. packet->header.type = CONNECTION_REQUEST;
  531. packet->header.credentials = credentials;
  532. packet->free = &outgoingPacketFree;
  533. }
  534. peerProxy = peerProxyCreate(peer, credentials);
  535. if (peerProxy == NULL) {
  536. free(packet);
  537. errno = ENOMEM;
  538. return NULL;
  539. } else {
  540. // Send a connection request to the master.
  541. if (requestConnection) {
  542. PeerProxy* masterProxy = peer->masterProxy;
  543. peerProxyEnqueueOutgoingPacket(masterProxy, packet);
  544. }
  545. return peerProxy;
  546. }
  547. }
  548. /**
  549. * Switches the master peer proxy into a state where it's waiting for a
  550. * connection from the master.
  551. */
  552. static void masterProxyExpectConnection(PeerProxy* masterProxy,
  553. Header* header) {
  554. // TODO: Restructure things so we don't need this check.
  555. // Verify that this really is the master.
  556. if (!masterProxy->master) {
  557. ALOGW("Non-master process %d tried to send us a connection.",
  558. masterProxy->credentials.pid);
  559. // Kill off the evil peer.
  560. peerProxyKill(masterProxy, false);
  561. return;
  562. }
  563. masterProxy->inputState = ACCEPTING_CONNECTION;
  564. Peer* localPeer = masterProxy->peer;
  565. // Create a peer proxy so we have somewhere to stash the creds.
  566. // See if we already have a proxy set up.
  567. pid_t pid = header->credentials.pid;
  568. peerLock(localPeer);
  569. PeerProxy* peerProxy = peerProxyGetOrCreate(localPeer, pid, false);
  570. if (peerProxy == NULL) {
  571. ALOGW("Peer proxy creation failed: %s", strerror(errno));
  572. } else {
  573. // Fill in full credentials.
  574. peerProxy->credentials = header->credentials;
  575. }
  576. peerUnlock(localPeer);
  577. // Keep track of which peer proxy we're accepting a connection for.
  578. masterProxy->connecting = peerProxy;
  579. }
  580. /**
  581. * Reads input from a peer process.
  582. */
  583. static void peerProxyRead(SelectableFd* fd);
  584. /** Sets up fd callbacks. */
  585. static void peerProxySetFd(PeerProxy* peerProxy, SelectableFd* fd) {
  586. peerProxy->fd = fd;
  587. fd->data = peerProxy;
  588. fd->onReadable = &peerProxyRead;
  589. fd->beforeSelect = &peerProxyBeforeSelect;
  590. // Make the socket non-blocking.
  591. setNonBlocking(fd->fd);
  592. }
  593. /**
  594. * Accepts a connection sent by the master proxy.
  595. */
  596. static void masterProxyAcceptConnection(PeerProxy* masterProxy) {
  597. struct msghdr msg;
  598. struct iovec iov[1];
  599. ssize_t size;
  600. char ignored;
  601. int incomingFd;
  602. // TODO: Reuse code which writes the connection. Who the heck designed
  603. // this API anyway?
  604. union {
  605. struct cmsghdr cm;
  606. char control[CMSG_SPACE(sizeof(int))];
  607. } control_un;
  608. struct cmsghdr *cmptr;
  609. msg.msg_control = control_un.control;
  610. msg.msg_controllen = sizeof(control_un.control);
  611. msg.msg_name = NULL;
  612. msg.msg_namelen = 0;
  613. // We sent 1 byte of data so we can detect EOF.
  614. iov[0].iov_base = &ignored;
  615. iov[0].iov_len = 1;
  616. msg.msg_iov = iov;
  617. msg.msg_iovlen = 1;
  618. size = recvmsg(masterProxy->fd->fd, &msg, 0);
  619. if (size < 0) {
  620. if (errno == EINTR) {
  621. // Log interruptions but otherwise ignore them.
  622. ALOGW("recvmsg() interrupted.");
  623. return;
  624. } else if (errno == EAGAIN) {
  625. // Keep waiting for the connection.
  626. return;
  627. } else {
  628. LOG_ALWAYS_FATAL("Error reading connection from master: %s",
  629. strerror(errno));
  630. }
  631. } else if (size == 0) {
  632. // EOF.
  633. LOG_ALWAYS_FATAL("Received EOF from master.");
  634. }
  635. // Extract fd from message.
  636. if ((cmptr = CMSG_FIRSTHDR(&msg)) != NULL
  637. && cmptr->cmsg_len == CMSG_LEN(sizeof(int))) {
  638. if (cmptr->cmsg_level != SOL_SOCKET) {
  639. LOG_ALWAYS_FATAL("Expected SOL_SOCKET.");
  640. }
  641. if (cmptr->cmsg_type != SCM_RIGHTS) {
  642. LOG_ALWAYS_FATAL("Expected SCM_RIGHTS.");
  643. }
  644. incomingFd = *((int*) CMSG_DATA(cmptr));
  645. } else {
  646. LOG_ALWAYS_FATAL("Expected fd.");
  647. }
  648. // The peer proxy this connection is for.
  649. PeerProxy* peerProxy = masterProxy->connecting;
  650. if (peerProxy == NULL) {
  651. ALOGW("Received connection for unknown peer.");
  652. closeWithWarning(incomingFd);
  653. } else {
  654. Peer* peer = masterProxy->peer;
  655. SelectableFd* selectableFd = selectorAdd(peer->selector, incomingFd);
  656. if (selectableFd == NULL) {
  657. ALOGW("Error adding fd to selector for %d.",
  658. peerProxy->credentials.pid);
  659. closeWithWarning(incomingFd);
  660. peerProxyKill(peerProxy, false);
  661. }
  662. peerProxySetFd(peerProxy, selectableFd);
  663. }
  664. peerProxyExpectHeader(masterProxy);
  665. }
  666. /**
  667. * Frees an outgoing packet containing a connection.
  668. */
  669. static void outgoingPacketFreeSocket(OutgoingPacket* packet) {
  670. closeWithWarning(packet->socket);
  671. outgoingPacketFree(packet);
  672. }
  673. /**
  674. * Connects two known peers.
  675. */
  676. static void masterConnectPeers(PeerProxy* peerA, PeerProxy* peerB) {
  677. int sockets[2];
  678. int result = socketpair(AF_LOCAL, SOCK_STREAM, 0, sockets);
  679. if (result == -1) {
  680. ALOGW("socketpair() error: %s", strerror(errno));
  681. // TODO: Send CONNECTION_FAILED packets to peers.
  682. return;
  683. }
  684. OutgoingPacket* packetA = calloc(1, sizeof(OutgoingPacket));
  685. OutgoingPacket* packetB = calloc(1, sizeof(OutgoingPacket));
  686. if (packetA == NULL || packetB == NULL) {
  687. free(packetA);
  688. free(packetB);
  689. ALOGW("malloc() error. Failed to tell process %d that process %d is"
  690. " dead.", peerA->credentials.pid, peerB->credentials.pid);
  691. return;
  692. }
  693. packetA->header.type = CONNECTION;
  694. packetB->header.type = CONNECTION;
  695. packetA->header.credentials = peerB->credentials;
  696. packetB->header.credentials = peerA->credentials;
  697. packetA->socket = sockets[0];
  698. packetB->socket = sockets[1];
  699. packetA->free = &outgoingPacketFreeSocket;
  700. packetB->free = &outgoingPacketFreeSocket;
  701. peerLock(peerA->peer);
  702. peerProxyEnqueueOutgoingPacket(peerA, packetA);
  703. peerProxyEnqueueOutgoingPacket(peerB, packetB);
  704. peerUnlock(peerA->peer);
  705. }
  706. /**
  707. * Informs a peer that the peer they're trying to connect to couldn't be
  708. * found.
  709. */
  710. static void masterReportConnectionError(PeerProxy* peerProxy,
  711. Credentials credentials) {
  712. OutgoingPacket* packet = calloc(1, sizeof(OutgoingPacket));
  713. if (packet == NULL) {
  714. ALOGW("malloc() error. Failed to tell process %d that process %d is"
  715. " dead.", peerProxy->credentials.pid, credentials.pid);
  716. return;
  717. }
  718. packet->header.type = CONNECTION_ERROR;
  719. packet->header.credentials = credentials;
  720. packet->free = &outgoingPacketFree;
  721. peerProxyLockAndEnqueueOutgoingPacket(peerProxy, packet);
  722. }
  723. /**
  724. * Handles a request to be connected to another peer.
  725. */
  726. static void masterHandleConnectionRequest(PeerProxy* peerProxy,
  727. Header* header) {
  728. Peer* master = peerProxy->peer;
  729. pid_t targetPid = header->credentials.pid;
  730. if (!hashmapContainsKey(peerProxy->connections, &targetPid)) {
  731. // We haven't connected these peers yet.
  732. PeerProxy* targetPeer
  733. = (PeerProxy*) hashmapGet(master->peerProxies, &targetPid);
  734. if (targetPeer == NULL) {
  735. // Unknown process.
  736. masterReportConnectionError(peerProxy, header->credentials);
  737. } else {
  738. masterConnectPeers(peerProxy, targetPeer);
  739. }
  740. }
  741. // This packet is complete. Get ready for the next one.
  742. peerProxyExpectHeader(peerProxy);
  743. }
  744. /**
  745. * The master told us this peer is dead.
  746. */
  747. static void masterProxyHandleConnectionError(PeerProxy* masterProxy,
  748. Header* header) {
  749. Peer* peer = masterProxy->peer;
  750. // Look up the peer proxy.
  751. pid_t pid = header->credentials.pid;
  752. PeerProxy* peerProxy = NULL;
  753. peerLock(peer);
  754. peerProxy = hashmapGet(peer->peerProxies, &pid);
  755. peerUnlock(peer);
  756. if (peerProxy != NULL) {
  757. ALOGI("Couldn't connect to %d.", pid);
  758. peerProxyKill(peerProxy, false);
  759. } else {
  760. ALOGW("Peer proxy for %d not found. This shouldn't happen.", pid);
  761. }
  762. peerProxyExpectHeader(masterProxy);
  763. }
  764. /**
  765. * Handles a packet header.
  766. */
  767. static void peerProxyHandleHeader(PeerProxy* peerProxy, Header* header) {
  768. switch (header->type) {
  769. case CONNECTION_REQUEST:
  770. masterHandleConnectionRequest(peerProxy, header);
  771. break;
  772. case CONNECTION:
  773. masterProxyExpectConnection(peerProxy, header);
  774. break;
  775. case CONNECTION_ERROR:
  776. masterProxyHandleConnectionError(peerProxy, header);
  777. break;
  778. case BYTES:
  779. peerProxyExpectBytes(peerProxy, header);
  780. break;
  781. default:
  782. ALOGW("Invalid packet type from %d: %d", peerProxy->credentials.pid,
  783. header->type);
  784. peerProxyKill(peerProxy, false);
  785. }
  786. }
  787. /**
  788. * Buffers input sent by peer. May be called multiple times until the entire
  789. * buffer is filled. Returns true when the buffer is full.
  790. */
  791. static bool peerProxyBufferInput(PeerProxy* peerProxy) {
  792. Buffer* in = peerProxy->inputBuffer;
  793. ssize_t size = bufferRead(in, peerProxy->fd->fd);
  794. if (size < 0) {
  795. peerProxyHandleError(peerProxy, "read");
  796. return false;
  797. } else if (size == 0) {
  798. // EOF.
  799. ALOGI("EOF");
  800. peerProxyKill(peerProxy, false);
  801. return false;
  802. } else if (bufferReadComplete(in)) {
  803. // We're done!
  804. return true;
  805. } else {
  806. // Continue reading.
  807. return false;
  808. }
  809. }
  810. /**
  811. * Reads input from a peer process.
  812. */
  813. static void peerProxyRead(SelectableFd* fd) {
  814. ALOGD("Reading...");
  815. PeerProxy* peerProxy = (PeerProxy*) fd->data;
  816. int state = peerProxy->inputState;
  817. Buffer* in = peerProxy->inputBuffer;
  818. switch (state) {
  819. case READING_HEADER:
  820. if (peerProxyBufferInput(peerProxy)) {
  821. ALOGD("Header read.");
  822. // We've read the complete header.
  823. Header* header = (Header*) in->data;
  824. peerProxyHandleHeader(peerProxy, header);
  825. }
  826. break;
  827. case READING_BYTES:
  828. ALOGD("Reading bytes...");
  829. if (peerProxyBufferInput(peerProxy)) {
  830. ALOGD("Bytes read.");
  831. // We have the complete packet. Notify bytes listener.
  832. peerProxy->peer->onBytes(peerProxy->credentials,
  833. in->data, in->size);
  834. // Get ready for the next packet.
  835. peerProxyExpectHeader(peerProxy);
  836. }
  837. break;
  838. case ACCEPTING_CONNECTION:
  839. masterProxyAcceptConnection(peerProxy);
  840. break;
  841. default:
  842. LOG_ALWAYS_FATAL("Unknown state: %d", state);
  843. }
  844. }
  845. static PeerProxy* peerProxyCreate(Peer* peer, Credentials credentials) {
  846. PeerProxy* peerProxy = calloc(1, sizeof(PeerProxy));
  847. if (peerProxy == NULL) {
  848. return NULL;
  849. }
  850. peerProxy->inputBuffer = bufferCreate(sizeof(Header));
  851. if (peerProxy->inputBuffer == NULL) {
  852. free(peerProxy);
  853. return NULL;
  854. }
  855. peerProxy->peer = peer;
  856. peerProxy->credentials = credentials;
  857. // Initial state == expecting a header.
  858. peerProxyExpectHeader(peerProxy);
  859. // Add this proxy to the map. Make sure the key points to the stable memory
  860. // inside of the peer proxy itself.
  861. pid_t* pid = &(peerProxy->credentials.pid);
  862. hashmapPut(peer->peerProxies, pid, peerProxy);
  863. return peerProxy;
  864. }
  865. /** Accepts a connection to the master peer. */
  866. static void masterAcceptConnection(SelectableFd* listenerFd) {
  867. // Accept connection.
  868. int socket = accept(listenerFd->fd, NULL, NULL);
  869. if (socket == -1) {
  870. ALOGW("accept() error: %s", strerror(errno));
  871. return;
  872. }
  873. ALOGD("Accepted connection as fd %d.", socket);
  874. // Get credentials.
  875. Credentials credentials;
  876. struct ucred ucredentials;
  877. socklen_t credentialsSize = sizeof(struct ucred);
  878. int result = getsockopt(socket, SOL_SOCKET, SO_PEERCRED,
  879. &ucredentials, &credentialsSize);
  880. // We might want to verify credentialsSize.
  881. if (result == -1) {
  882. ALOGW("getsockopt() error: %s", strerror(errno));
  883. closeWithWarning(socket);
  884. return;
  885. }
  886. // Copy values into our own structure so we know we have the types right.
  887. credentials.pid = ucredentials.pid;
  888. credentials.uid = ucredentials.uid;
  889. credentials.gid = ucredentials.gid;
  890. ALOGI("Accepted connection from process %d.", credentials.pid);
  891. Peer* masterPeer = (Peer*) listenerFd->data;
  892. peerLock(masterPeer);
  893. // Make sure we don't already have a connection from that process.
  894. PeerProxy* peerProxy
  895. = hashmapGet(masterPeer->peerProxies, &credentials.pid);
  896. if (peerProxy != NULL) {
  897. peerUnlock(masterPeer);
  898. ALOGW("Alread connected to process %d.", credentials.pid);
  899. closeWithWarning(socket);
  900. return;
  901. }
  902. // Add connection to the selector.
  903. SelectableFd* socketFd = selectorAdd(masterPeer->selector, socket);
  904. if (socketFd == NULL) {
  905. peerUnlock(masterPeer);
  906. ALOGW("malloc() failed.");
  907. closeWithWarning(socket);
  908. return;
  909. }
  910. // Create a peer proxy.
  911. peerProxy = peerProxyCreate(masterPeer, credentials);
  912. peerUnlock(masterPeer);
  913. if (peerProxy == NULL) {
  914. ALOGW("malloc() failed.");
  915. socketFd->remove = true;
  916. closeWithWarning(socket);
  917. }
  918. peerProxy->connections = hashmapCreate(10, &pidHash, &pidEquals);
  919. peerProxySetFd(peerProxy, socketFd);
  920. }
  921. /**
  922. * Creates the local peer.
  923. */
  924. static Peer* peerCreate() {
  925. Peer* peer = calloc(1, sizeof(Peer));
  926. if (peer == NULL) {
  927. LOG_ALWAYS_FATAL("malloc() error.");
  928. }
  929. peer->peerProxies = hashmapCreate(10, &pidHash, &pidEquals);
  930. peer->selector = selectorCreate();
  931. pthread_mutexattr_t attributes;
  932. if (pthread_mutexattr_init(&attributes) != 0) {
  933. LOG_ALWAYS_FATAL("pthread_mutexattr_init() error.");
  934. }
  935. if (pthread_mutexattr_settype(&attributes, PTHREAD_MUTEX_RECURSIVE) != 0) {
  936. LOG_ALWAYS_FATAL("pthread_mutexattr_settype() error.");
  937. }
  938. if (pthread_mutex_init(&peer->mutex, &attributes) != 0) {
  939. LOG_ALWAYS_FATAL("pthread_mutex_init() error.");
  940. }
  941. peer->pid = getpid();
  942. return peer;
  943. }
  944. /** The local peer. */
  945. static Peer* localPeer;
  946. /** Frees a packet of bytes. */
  947. static void outgoingPacketFreeBytes(OutgoingPacket* packet) {
  948. ALOGD("Freeing outgoing packet.");
  949. bufferFree(packet->bytes);
  950. free(packet);
  951. }
  952. /**
  953. * Sends a packet of bytes to a remote peer. Returns 0 on success.
  954. *
  955. * Returns -1 if an error occurs. Sets errno to ENOMEM if memory couldn't be
  956. * allocated. Sets errno to EHOSTDOWN if the peer died recently. Sets errno
  957. * to EINVAL if pid is the same as the local pid.
  958. */
  959. int peerSendBytes(pid_t pid, const char* bytes, size_t size) {
  960. Peer* peer = localPeer;
  961. assert(peer != NULL);
  962. OutgoingPacket* packet = calloc(1, sizeof(OutgoingPacket));
  963. if (packet == NULL) {
  964. errno = ENOMEM;
  965. return -1;
  966. }
  967. Buffer* copy = bufferCreate(size);
  968. if (copy == NULL) {
  969. free(packet);
  970. errno = ENOMEM;
  971. return -1;
  972. }
  973. // Copy data.
  974. memcpy(copy->data, bytes, size);
  975. copy->size = size;
  976. packet->bytes = copy;
  977. packet->header.type = BYTES;
  978. packet->header.size = size;
  979. packet->free = outgoingPacketFreeBytes;
  980. bufferPrepareForWrite(packet->bytes);
  981. peerLock(peer);
  982. PeerProxy* peerProxy = peerProxyGetOrCreate(peer, pid, true);
  983. if (peerProxy == NULL) {
  984. // The peer is already dead or we couldn't alloc memory. Either way,
  985. // errno is set.
  986. peerUnlock(peer);
  987. packet->free(packet);
  988. return -1;
  989. } else {
  990. peerProxyEnqueueOutgoingPacket(peerProxy, packet);
  991. peerUnlock(peer);
  992. selectorWakeUp(peer->selector);
  993. return 0;
  994. }
  995. }
  996. /** Keeps track of how to free shared bytes. */
  997. typedef struct {
  998. void (*free)(void* context);
  999. void* context;
  1000. } SharedBytesFreer;
  1001. /** Frees shared bytes. */
  1002. static void outgoingPacketFreeSharedBytes(OutgoingPacket* packet) {
  1003. SharedBytesFreer* sharedBytesFreer
  1004. = (SharedBytesFreer*) packet->context;
  1005. sharedBytesFreer->free(sharedBytesFreer->context);
  1006. free(sharedBytesFreer);
  1007. free(packet);
  1008. }
  1009. /**
  1010. * Sends a packet of bytes to a remote peer without copying the bytes. Calls
  1011. * free() with context after the bytes have been sent.
  1012. *
  1013. * Returns -1 if an error occurs. Sets errno to ENOMEM if memory couldn't be
  1014. * allocated. Sets errno to EHOSTDOWN if the peer died recently. Sets errno
  1015. * to EINVAL if pid is the same as the local pid.
  1016. */
  1017. int peerSendSharedBytes(pid_t pid, char* bytes, size_t size,
  1018. void (*free)(void* context), void* context) {
  1019. Peer* peer = localPeer;
  1020. assert(peer != NULL);
  1021. OutgoingPacket* packet = calloc(1, sizeof(OutgoingPacket));
  1022. if (packet == NULL) {
  1023. errno = ENOMEM;
  1024. return -1;
  1025. }
  1026. Buffer* wrapper = bufferWrap(bytes, size, size);
  1027. if (wrapper == NULL) {
  1028. free(packet);
  1029. errno = ENOMEM;
  1030. return -1;
  1031. }
  1032. SharedBytesFreer* sharedBytesFreer = malloc(sizeof(SharedBytesFreer));
  1033. if (sharedBytesFreer == NULL) {
  1034. free(packet);
  1035. free(wrapper);
  1036. errno = ENOMEM;
  1037. return -1;
  1038. }
  1039. sharedBytesFreer->free = free;
  1040. sharedBytesFreer->context = context;
  1041. packet->bytes = wrapper;
  1042. packet->context = sharedBytesFreer;
  1043. packet->header.type = BYTES;
  1044. packet->header.size = size;
  1045. packet->free = &outgoingPacketFreeSharedBytes;
  1046. bufferPrepareForWrite(packet->bytes);
  1047. peerLock(peer);
  1048. PeerProxy* peerProxy = peerProxyGetOrCreate(peer, pid, true);
  1049. if (peerProxy == NULL) {
  1050. // The peer is already dead or we couldn't alloc memory. Either way,
  1051. // errno is set.
  1052. peerUnlock(peer);
  1053. packet->free(packet);
  1054. return -1;
  1055. } else {
  1056. peerProxyEnqueueOutgoingPacket(peerProxy, packet);
  1057. peerUnlock(peer);
  1058. selectorWakeUp(peer->selector);
  1059. return 0;
  1060. }
  1061. }
  1062. /**
  1063. * Starts the master peer. The master peer differs from other peers in that
  1064. * it is responsible for connecting the other peers. You can only have one
  1065. * master peer.
  1066. *
  1067. * Goes into an I/O loop and does not return.
  1068. */
  1069. void masterPeerInitialize(BytesListener* bytesListener,
  1070. DeathListener* deathListener) {
  1071. // Create and bind socket.
  1072. int listenerSocket = socket(AF_LOCAL, SOCK_STREAM, 0);
  1073. if (listenerSocket == -1) {
  1074. LOG_ALWAYS_FATAL("socket() error: %s", strerror(errno));
  1075. }
  1076. unlink(MASTER_PATH);
  1077. int result = bind(listenerSocket, (SocketAddress*) getMasterAddress(),
  1078. sizeof(UnixAddress));
  1079. if (result == -1) {
  1080. LOG_ALWAYS_FATAL("bind() error: %s", strerror(errno));
  1081. }
  1082. ALOGD("Listener socket: %d", listenerSocket);
  1083. // Queue up to 16 connections.
  1084. result = listen(listenerSocket, 16);
  1085. if (result != 0) {
  1086. LOG_ALWAYS_FATAL("listen() error: %s", strerror(errno));
  1087. }
  1088. // Make socket non-blocking.
  1089. setNonBlocking(listenerSocket);
  1090. // Create the peer for this process. Fail if we already have one.
  1091. if (localPeer != NULL) {
  1092. LOG_ALWAYS_FATAL("Peer is already initialized.");
  1093. }
  1094. localPeer = peerCreate();
  1095. if (localPeer == NULL) {
  1096. LOG_ALWAYS_FATAL("malloc() failed.");
  1097. }
  1098. localPeer->master = true;
  1099. localPeer->onBytes = bytesListener;
  1100. localPeer->onDeath = deathListener;
  1101. // Make listener socket selectable.
  1102. SelectableFd* listenerFd = selectorAdd(localPeer->selector, listenerSocket);
  1103. if (listenerFd == NULL) {
  1104. LOG_ALWAYS_FATAL("malloc() error.");
  1105. }
  1106. listenerFd->data = localPeer;
  1107. listenerFd->onReadable = &masterAcceptConnection;
  1108. }
  1109. /**
  1110. * Starts a local peer.
  1111. *
  1112. * Goes into an I/O loop and does not return.
  1113. */
  1114. void peerInitialize(BytesListener* bytesListener,
  1115. DeathListener* deathListener) {
  1116. // Connect to master peer.
  1117. int masterSocket = socket(AF_LOCAL, SOCK_STREAM, 0);
  1118. if (masterSocket == -1) {
  1119. LOG_ALWAYS_FATAL("socket() error: %s", strerror(errno));
  1120. }
  1121. int result = connect(masterSocket, (SocketAddress*) getMasterAddress(),
  1122. sizeof(UnixAddress));
  1123. if (result != 0) {
  1124. LOG_ALWAYS_FATAL("connect() error: %s", strerror(errno));
  1125. }
  1126. // Create the peer for this process. Fail if we already have one.
  1127. if (localPeer != NULL) {
  1128. LOG_ALWAYS_FATAL("Peer is already initialized.");
  1129. }
  1130. localPeer = peerCreate();
  1131. if (localPeer == NULL) {
  1132. LOG_ALWAYS_FATAL("malloc() failed.");
  1133. }
  1134. localPeer->onBytes = bytesListener;
  1135. localPeer->onDeath = deathListener;
  1136. // Make connection selectable.
  1137. SelectableFd* masterFd = selectorAdd(localPeer->selector, masterSocket);
  1138. if (masterFd == NULL) {
  1139. LOG_ALWAYS_FATAL("malloc() error.");
  1140. }
  1141. // Create a peer proxy for the master peer.
  1142. PeerProxy* masterProxy = peerProxyCreate(localPeer, MASTER_CREDENTIALS);
  1143. if (masterProxy == NULL) {
  1144. LOG_ALWAYS_FATAL("malloc() error.");
  1145. }
  1146. peerProxySetFd(masterProxy, masterFd);
  1147. masterProxy->master = true;
  1148. localPeer->masterProxy = masterProxy;
  1149. }
  1150. /** Starts the master peer I/O loop. Doesn't return. */
  1151. void peerLoop() {
  1152. assert(localPeer != NULL);
  1153. // Start selector.
  1154. selectorLoop(localPeer->selector);
  1155. }