CernVM-FS  2.9.0
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
cache_transport.cc
Go to the documentation of this file.
1 
4 #include "cvmfs_config.h"
5 #include "cache_transport.h"
6 
7 #include <alloca.h>
8 #include <errno.h>
9 #include <sys/socket.h>
10 
11 #include <cassert>
12 #include <cstdlib>
13 #include <cstring>
14 
15 #include "hash.h"
16 #include "logging.h"
17 #include "smalloc.h"
18 #include "util/exception.h"
19 #include "util/posix.h"
20 
21 // TODO(jblomer): Check for possible starvation of plugin by dying clients
22 // (blocking read). Probably only relevant for TCP sockets.
23 
24 using namespace std; // NOLINT
25 
27  "__CVMFS_CACHE_EXTERNAL_PIPE_READY__";
28 
34  assert(msg_typed_ != NULL);
35  if (!is_wrapped_)
36  WrapMsg();
37  return &msg_rpc_;
38 }
39 
40 
45 google::protobuf::MessageLite *CacheTransport::Frame::GetMsgTyped() {
46  assert(msg_rpc_.IsInitialized());
47  if (msg_typed_ == NULL)
48  UnwrapMsg();
49  return msg_typed_;
50 }
51 
52 
54  : owns_msg_typed_(false)
55  , msg_typed_(NULL)
56  , attachment_(NULL)
57  , att_size_(0)
58  , is_wrapped_(false)
59  , is_msg_out_of_band_(false)
60 { }
61 
62 
63 CacheTransport::Frame::Frame(google::protobuf::MessageLite *m)
64  : owns_msg_typed_(false)
65  , msg_typed_(m)
66  , attachment_(NULL)
67  , att_size_(0)
68  , is_wrapped_(false)
69  , is_msg_out_of_band_(false)
70 { }
71 
72 
74  Reset(0);
75 }
76 
77 
79  assert(msg_rpc_.IsInitialized());
80  if (msg_typed_ == NULL)
81  UnwrapMsg();
82  return is_msg_out_of_band_;
83 }
84 
85 
87  msg_rpc_.CheckTypeAndMergeFrom(other.msg_rpc_);
88  owns_msg_typed_ = true;
89  if (other.att_size_ > 0) {
90  assert(att_size_ >= other.att_size_);
91  memcpy(attachment_, other.attachment_, other.att_size_);
92  att_size_ = other.att_size_;
93  }
94 }
95 
96 
97 bool CacheTransport::Frame::ParseMsgRpc(void *buffer, uint32_t size) {
98  bool retval = msg_rpc_.ParseFromArray(buffer, size);
99  if (!retval)
100  return false;
101 
102  // Cleanup typed message when Frame leaves scope
103  owns_msg_typed_ = true;
104  return true;
105 }
106 
107 
109  if (owns_msg_typed_)
110  return;
111 
112  msg_rpc_.release_msg_refcount_req();
113  msg_rpc_.release_msg_refcount_reply();
114  msg_rpc_.release_msg_read_req();
115  msg_rpc_.release_msg_read_reply();
116  msg_rpc_.release_msg_object_info_req();
117  msg_rpc_.release_msg_object_info_reply();
118  msg_rpc_.release_msg_store_req();
119  msg_rpc_.release_msg_store_abort_req();
120  msg_rpc_.release_msg_store_reply();
121  msg_rpc_.release_msg_handshake();
122  msg_rpc_.release_msg_handshake_ack();
123  msg_rpc_.release_msg_quit();
124  msg_rpc_.release_msg_ioctl();
125  msg_rpc_.release_msg_info_req();
126  msg_rpc_.release_msg_info_reply();
127  msg_rpc_.release_msg_shrink_req();
128  msg_rpc_.release_msg_shrink_reply();
129  msg_rpc_.release_msg_list_req();
130  msg_rpc_.release_msg_list_reply();
131  msg_rpc_.release_msg_detach();
132  msg_rpc_.release_msg_breadcrumb_store_req();
133  msg_rpc_.release_msg_breadcrumb_load_req();
134  msg_rpc_.release_msg_breadcrumb_reply();
135 }
136 
137 
138 void CacheTransport::Frame::Reset(uint32_t original_att_size) {
139  msg_typed_ = NULL;
140  att_size_ = original_att_size;
141  is_wrapped_ = false;
142  is_msg_out_of_band_ = false;
143  Release();
144  msg_rpc_.Clear();
145  owns_msg_typed_ = false;
146 }
147 
148 
150  if (msg_typed_->GetTypeName() == "cvmfs.MsgHandshake") {
151  msg_rpc_.set_allocated_msg_handshake(
152  reinterpret_cast<cvmfs::MsgHandshake *>(msg_typed_));
153  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgHandshakeAck") {
154  msg_rpc_.set_allocated_msg_handshake_ack(
155  reinterpret_cast<cvmfs::MsgHandshakeAck *>(msg_typed_));
156  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgQuit") {
157  msg_rpc_.set_allocated_msg_quit(
158  reinterpret_cast<cvmfs::MsgQuit *>(msg_typed_));
159  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgIoctl") {
160  msg_rpc_.set_allocated_msg_ioctl(
161  reinterpret_cast<cvmfs::MsgIoctl *>(msg_typed_));
162  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgRefcountReq") {
163  msg_rpc_.set_allocated_msg_refcount_req(
164  reinterpret_cast<cvmfs::MsgRefcountReq *>(msg_typed_));
165  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgRefcountReply") {
166  msg_rpc_.set_allocated_msg_refcount_reply(
167  reinterpret_cast<cvmfs::MsgRefcountReply *>(msg_typed_));
168  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgObjectInfoReq") {
169  msg_rpc_.set_allocated_msg_object_info_req(
170  reinterpret_cast<cvmfs::MsgObjectInfoReq *>(msg_typed_));
171  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgObjectInfoReply") {
172  msg_rpc_.set_allocated_msg_object_info_reply(
173  reinterpret_cast<cvmfs::MsgObjectInfoReply *>(msg_typed_));
174  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgReadReq") {
175  msg_rpc_.set_allocated_msg_read_req(
176  reinterpret_cast<cvmfs::MsgReadReq *>(msg_typed_));
177  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgReadReply") {
178  msg_rpc_.set_allocated_msg_read_reply(
179  reinterpret_cast<cvmfs::MsgReadReply *>(msg_typed_));
180  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgStoreReq") {
181  msg_rpc_.set_allocated_msg_store_req(
182  reinterpret_cast<cvmfs::MsgStoreReq *>(msg_typed_));
183  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgStoreAbortReq") {
184  msg_rpc_.set_allocated_msg_store_abort_req(
185  reinterpret_cast<cvmfs::MsgStoreAbortReq *>(msg_typed_));
186  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgStoreReply") {
187  msg_rpc_.set_allocated_msg_store_reply(
188  reinterpret_cast<cvmfs::MsgStoreReply *>(msg_typed_));
189  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgInfoReq") {
190  msg_rpc_.set_allocated_msg_info_req(
191  reinterpret_cast<cvmfs::MsgInfoReq *>(msg_typed_));
192  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgInfoReply") {
193  msg_rpc_.set_allocated_msg_info_reply(
194  reinterpret_cast<cvmfs::MsgInfoReply *>(msg_typed_));
195  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgShrinkReq") {
196  msg_rpc_.set_allocated_msg_shrink_req(
197  reinterpret_cast<cvmfs::MsgShrinkReq *>(msg_typed_));
198  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgShrinkReply") {
199  msg_rpc_.set_allocated_msg_shrink_reply(
200  reinterpret_cast<cvmfs::MsgShrinkReply *>(msg_typed_));
201  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgListReq") {
202  msg_rpc_.set_allocated_msg_list_req(
203  reinterpret_cast<cvmfs::MsgListReq *>(msg_typed_));
204  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgListReply") {
205  msg_rpc_.set_allocated_msg_list_reply(
206  reinterpret_cast<cvmfs::MsgListReply *>(msg_typed_));
207  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgBreadcrumbStoreReq") {
208  msg_rpc_.set_allocated_msg_breadcrumb_store_req(
209  reinterpret_cast<cvmfs::MsgBreadcrumbStoreReq *>(msg_typed_));
210  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgBreadcrumbLoadReq") {
211  msg_rpc_.set_allocated_msg_breadcrumb_load_req(
212  reinterpret_cast<cvmfs::MsgBreadcrumbLoadReq *>(msg_typed_));
213  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgBreadcrumbReply") {
214  msg_rpc_.set_allocated_msg_breadcrumb_reply(
215  reinterpret_cast<cvmfs::MsgBreadcrumbReply *>(msg_typed_));
216  } else if (msg_typed_->GetTypeName() == "cvmfs.MsgDetach") {
217  msg_rpc_.set_allocated_msg_detach(
218  reinterpret_cast<cvmfs::MsgDetach *>(msg_typed_));
219  is_msg_out_of_band_ = true;
220  } else {
221  // Unexpected message type, should never happen
222  PANIC(NULL);
223  }
224  is_wrapped_ = true;
225 }
226 
227 
229  if (msg_rpc_.has_msg_handshake()) {
230  msg_typed_ = msg_rpc_.mutable_msg_handshake();
231  } else if (msg_rpc_.has_msg_handshake_ack()) {
232  msg_typed_ = msg_rpc_.mutable_msg_handshake_ack();
233  } else if (msg_rpc_.has_msg_quit()) {
234  msg_typed_ = msg_rpc_.mutable_msg_quit();
235  } else if (msg_rpc_.has_msg_ioctl()) {
236  msg_typed_ = msg_rpc_.mutable_msg_ioctl();
237  } else if (msg_rpc_.has_msg_refcount_req()) {
238  msg_typed_ = msg_rpc_.mutable_msg_refcount_req();
239  } else if (msg_rpc_.has_msg_refcount_reply()) {
240  msg_typed_ = msg_rpc_.mutable_msg_refcount_reply();
241  } else if (msg_rpc_.has_msg_object_info_req()) {
242  msg_typed_ = msg_rpc_.mutable_msg_object_info_req();
243  } else if (msg_rpc_.has_msg_object_info_reply()) {
244  msg_typed_ = msg_rpc_.mutable_msg_object_info_reply();
245  } else if (msg_rpc_.has_msg_read_req()) {
246  msg_typed_ = msg_rpc_.mutable_msg_read_req();
247  } else if (msg_rpc_.has_msg_read_reply()) {
248  msg_typed_ = msg_rpc_.mutable_msg_read_reply();
249  } else if (msg_rpc_.has_msg_store_req()) {
250  msg_typed_ = msg_rpc_.mutable_msg_store_req();
251  } else if (msg_rpc_.has_msg_store_abort_req()) {
252  msg_typed_ = msg_rpc_.mutable_msg_store_abort_req();
253  } else if (msg_rpc_.has_msg_store_reply()) {
254  msg_typed_ = msg_rpc_.mutable_msg_store_reply();
255  } else if (msg_rpc_.has_msg_info_req()) {
256  msg_typed_ = msg_rpc_.mutable_msg_info_req();
257  } else if (msg_rpc_.has_msg_info_reply()) {
258  msg_typed_ = msg_rpc_.mutable_msg_info_reply();
259  } else if (msg_rpc_.has_msg_shrink_req()) {
260  msg_typed_ = msg_rpc_.mutable_msg_shrink_req();
261  } else if (msg_rpc_.has_msg_shrink_reply()) {
262  msg_typed_ = msg_rpc_.mutable_msg_shrink_reply();
263  } else if (msg_rpc_.has_msg_list_req()) {
264  msg_typed_ = msg_rpc_.mutable_msg_list_req();
265  } else if (msg_rpc_.has_msg_list_reply()) {
266  msg_typed_ = msg_rpc_.mutable_msg_list_reply();
267  } else if (msg_rpc_.has_msg_breadcrumb_store_req()) {
268  msg_typed_ = msg_rpc_.mutable_msg_breadcrumb_store_req();
269  } else if (msg_rpc_.has_msg_breadcrumb_load_req()) {
270  msg_typed_ = msg_rpc_.mutable_msg_breadcrumb_load_req();
271  } else if (msg_rpc_.has_msg_breadcrumb_reply()) {
272  msg_typed_ = msg_rpc_.mutable_msg_breadcrumb_reply();
273  } else if (msg_rpc_.has_msg_detach()) {
274  msg_typed_ = msg_rpc_.mutable_msg_detach();
275  is_msg_out_of_band_ = true;
276  } else {
277  // Unexpected message type, should never happen
278  PANIC(NULL);
279  }
280 }
281 
282 
283 //------------------------------------------------------------------------------
284 
285 
287  : fd_connection_(fd_connection)
288  , flags_(0)
289 {
290  assert(fd_connection_ >= 0);
291 }
292 
293 
294 CacheTransport::CacheTransport(int fd_connection, uint32_t flags)
295  : fd_connection_(fd_connection)
296  , flags_(flags)
297 {
298  assert(fd_connection_ >= 0);
299 }
300 
301 
303  const shash::Any &hash,
304  cvmfs::MsgHash *msg_hash)
305 {
306  switch (hash.algorithm) {
307  case shash::kSha1:
308  msg_hash->set_algorithm(cvmfs::HASH_SHA1);
309  break;
310  case shash::kRmd160:
311  msg_hash->set_algorithm(cvmfs::HASH_RIPEMD160);
312  break;
313  case shash::kShake128:
314  msg_hash->set_algorithm(cvmfs::HASH_SHAKE128);
315  break;
316  default:
317  PANIC(NULL);
318  }
319  msg_hash->set_digest(hash.digest, shash::kDigestSizes[hash.algorithm]);
320 }
321 
322 
324  CacheManager::ObjectType object_type,
325  cvmfs::EnumObjectType *wire_type)
326 {
327  switch (object_type) {
329  // TODO(jblomer): "pinned" should mean a permanently open fd
331  *wire_type = cvmfs::OBJECT_REGULAR;
332  break;
334  *wire_type = cvmfs::OBJECT_CATALOG;
335  break;
337  *wire_type = cvmfs::OBJECT_VOLATILE;
338  break;
339  default:
340  PANIC(NULL);
341  }
342 }
343 
344 
346  const cvmfs::MsgHash &msg_hash,
347  shash::Any *hash)
348 {
349  switch (msg_hash.algorithm()) {
350  case cvmfs::HASH_SHA1:
351  hash->algorithm = shash::kSha1;
352  break;
353  case cvmfs::HASH_RIPEMD160:
354  hash->algorithm = shash::kRmd160;
355  break;
356  case cvmfs::HASH_SHAKE128:
357  hash->algorithm = shash::kShake128;
358  break;
359  default:
360  return false;
361  }
362  const unsigned digest_size = shash::kDigestSizes[hash->algorithm];
363  if (msg_hash.digest().length() != digest_size)
364  return false;
365  memcpy(hash->digest, msg_hash.digest().data(), digest_size);
366  return true;
367 }
368 
369 
371  cvmfs::EnumObjectType wire_type,
372  CacheManager::ObjectType *object_type)
373 {
374  switch (wire_type) {
375  case cvmfs::OBJECT_REGULAR:
376  *object_type = CacheManager::kTypeRegular;
377  return true;
378  case cvmfs::OBJECT_CATALOG:
379  *object_type = CacheManager::kTypeCatalog;
380  return true;
381  case cvmfs::OBJECT_VOLATILE:
382  *object_type = CacheManager::kTypeVolatile;
383  return true;
384  default:
385  return false;
386  }
387 }
388 
389 
391  uint32_t size;
392  bool has_attachment;
393  bool retval = RecvHeader(&size, &has_attachment);
394  if (!retval)
395  return false;
396 
397  void *buffer;
398  if (size <= kMaxStackAlloc)
399  buffer = alloca(size);
400  else
401  buffer = smalloc(size);
402  ssize_t nbytes = SafeRead(fd_connection_, buffer, size);
403  if ((nbytes < 0) || (static_cast<uint32_t>(nbytes) != size)) {
404  if (size > kMaxStackAlloc) { free(buffer); }
405  return false;
406  }
407 
408  uint32_t msg_size = size;
409  if (has_attachment) {
410  if (size < 2) {
411  // kMaxStackAlloc is > 2 (of course!) but we'll leave the condition here
412  // for consistency.
413  if (size > kMaxStackAlloc) { free(buffer); }
414  return false;
415  }
416  msg_size = (*reinterpret_cast<unsigned char *>(buffer)) +
417  ((*(reinterpret_cast<unsigned char *>(buffer) + 1)) << 8);
418  if ((msg_size + kInnerHeaderSize) > size) {
419  if (size > kMaxStackAlloc) { free(buffer); }
420  return false;
421  }
422  }
423 
424  void *ptr_msg = has_attachment
425  ? (reinterpret_cast<char *>(buffer) + kInnerHeaderSize)
426  : buffer;
427  retval = frame->ParseMsgRpc(ptr_msg, msg_size);
428  if (!retval) {
429  if (size > kMaxStackAlloc) { free(buffer); }
430  return false;
431  }
432 
433  if (has_attachment) {
434  uint32_t attachment_size = size - (msg_size + kInnerHeaderSize);
435  if (frame->att_size() < attachment_size) {
436  if (size > kMaxStackAlloc) { free(buffer); }
437  return false;
438  }
439  void *ptr_attachment =
440  reinterpret_cast<char *>(buffer) + kInnerHeaderSize + msg_size;
441  memcpy(frame->attachment(), ptr_attachment, attachment_size);
442  frame->set_att_size(attachment_size);
443  } else {
444  frame->set_att_size(0);
445  }
446  if (size > kMaxStackAlloc) { free(buffer); }
447  return true;
448 }
449 
450 
451 bool CacheTransport::RecvHeader(uint32_t *size, bool *has_attachment) {
452  unsigned char header[kHeaderSize];
453  ssize_t nbytes = SafeRead(fd_connection_, header, kHeaderSize);
454  if ((nbytes < 0) || (static_cast<unsigned>(nbytes) != kHeaderSize))
455  return false;
456  if ((header[0] & (~kFlagHasAttachment)) != kWireProtocolVersion)
457  return false;
458  *has_attachment = header[0] & kFlagHasAttachment;
459  *size = header[1] + (header[2] << 8) + (header[3] << 16);
460  return (*size > 0) && (*size <= kMaxMsgSize);
461 }
462 
463 
465  void *message,
466  uint32_t msg_size,
467  void *attachment,
468  uint32_t att_size)
469 {
470  uint32_t total_size =
471  msg_size + att_size + ((att_size > 0) ? kInnerHeaderSize : 0);
472 
473  assert(total_size > 0);
474  assert(total_size <= kMaxMsgSize);
476  "sending message of size %u to cache transport", total_size);
477 
478  unsigned char header[kHeaderSize];
479  header[0] = kWireProtocolVersion | ((att_size == 0) ? 0 : kFlagHasAttachment);
480  header[1] = (total_size & 0x000000FF);
481  header[2] = (total_size & 0x0000FF00) >> 8;
482  header[3] = (total_size & 0x00FF0000) >> 16;
483  // Only transferred if an attachment is present. Otherwise the overall size
484  // is also the size of the protobuf message.
485  unsigned char inner_header[kInnerHeaderSize];
486 
487  struct iovec iov[4];
488  iov[0].iov_base = header;
489  iov[0].iov_len = kHeaderSize;
490 
491  if (att_size > 0) {
492  inner_header[0] = (msg_size & 0x000000FF);
493  inner_header[1] = (msg_size & 0x0000FF00) >> 8;
494  iov[1].iov_base = inner_header;
495  iov[1].iov_len = kInnerHeaderSize;
496  iov[2].iov_base = message;
497  iov[2].iov_len = msg_size;
498  iov[3].iov_base = attachment;
499  iov[3].iov_len = att_size;
500  } else {
501  iov[1].iov_base = message;
502  iov[1].iov_len = msg_size;
503  }
505  SendNonBlocking(iov, (att_size == 0) ? 2 : 4);
506  return;
507  }
508  bool retval = SafeWriteV(fd_connection_, iov, (att_size == 0) ? 2 : 4);
509 
510  if (!retval && !(flags_ & kFlagSendIgnoreFailure)) {
512  "failed to write to external cache transport (%d), aborting", errno);
513  }
514 }
515 
516 void CacheTransport::SendNonBlocking(struct iovec *iov, unsigned iovcnt) {
517  assert(iovcnt > 0);
518  unsigned total_size = 0;
519  for (unsigned i = 0; i < iovcnt; ++i)
520  total_size += iov[i].iov_len;
521  unsigned char *buffer = reinterpret_cast<unsigned char *>(alloca(total_size));
522 
523  unsigned pos = 0;
524  for (unsigned i = 0; i < iovcnt; ++i) {
525  memcpy(buffer + pos, iov[i].iov_base, iov[i].iov_len);
526  pos += iov[i].iov_len;
527  }
528 
529  int retval = send(fd_connection_, buffer, total_size, MSG_DONTWAIT);
530  if (retval < 0) {
531  assert(errno != EMSGSIZE);
532  if (!(flags_ & kFlagSendIgnoreFailure)) {
534  "failed to write to external cache transport (%d), aborting",
535  errno);
536  }
537  }
538 }
539 
540 
542  cvmfs::MsgRpc *msg_rpc = frame->GetMsgRpc();
543  int32_t size = msg_rpc->ByteSize();
544  assert(size > 0);
545 #ifdef __APPLE__
546  void *buffer = smalloc(size);
547 #else
548  void *buffer = alloca(size);
549 #endif
550  bool retval = msg_rpc->SerializeToArray(buffer, size);
551  assert(retval);
552  SendData(buffer, size, frame->attachment(), frame->att_size());
553 #ifdef __APPLE__
554  free(buffer);
555 #endif
556 }
#define LogCvmfs(source, mask,...)
Definition: logging.h:20
static const unsigned kMaxStackAlloc
#define PANIC(...)
Definition: exception.h:26
static const unsigned char kWireProtocolVersion
void SendNonBlocking(struct iovec *iov, unsigned iovcnt)
assert((mem||(size==0))&&"Out Of Memory")
void SendFrame(Frame *frame)
Algorithms algorithm
Definition: hash.h:123
unsigned char digest[digest_size_]
Definition: hash.h:122
static const uint32_t kFlagSendIgnoreFailure
void FillMsgHash(const shash::Any &hash, cvmfs::MsgHash *msg_hash)
void MergeFrom(const Frame &other)
static const uint32_t kMaxMsgSize
ssize_t SafeRead(int fd, void *buf, size_t nbyte)
Definition: posix.cc:1983
void * attachment() const
bool ParseMsgHash(const cvmfs::MsgHash &msg_hash, shash::Any *hash)
cvmfs::MsgRpc * GetMsgRpc()
uint32_t att_size() const
void set_att_size(uint32_t size)
static const unsigned char kFlagHasAttachment
google::protobuf::MessageLite * GetMsgTyped()
void SendData(void *message, uint32_t msg_size, void *attachment=NULL, uint32_t att_size=0)
bool ParseMsgRpc(void *buffer, uint32_t size)
bool ParseObjectType(cvmfs::EnumObjectType wire_type, CacheManager::ObjectType *object_type)
bool RecvFrame(Frame *frame)
static const unsigned kInnerHeaderSize
void Reset(uint32_t original_att_size)
static void size_t size
Definition: smalloc.h:47
void FillObjectType(CacheManager::ObjectType object_type, cvmfs::EnumObjectType *wire_type)
const unsigned kDigestSizes[]
Definition: hash.h:67
static const unsigned kHeaderSize
int fd_connection() const
bool SafeWriteV(int fd, struct iovec *iov, unsigned iovcnt)
Definition: posix.cc:1942
static const char * kEnvReadyNotifyFd
static const uint32_t kFlagSendNonBlocking
CacheTransport(int fd_connection)
bool RecvHeader(uint32_t *size, bool *has_attachment)