mojo/system/data_pipe.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "mojo/system/data_pipe.h"
   6
   7 #include <string.h>
   8
   9 #include <algorithm>
  10 #include <limits>
  11
  12 #include "base/compiler_specific.h"
  13 #include "base/logging.h"
  14 #include "mojo/system/constants.h"
  15 #include "mojo/system/memory.h"
  16 #include "mojo/system/options_validation.h"
  17 #include "mojo/system/waiter_list.h"
  18
  19 namespace mojo {
  20 namespace system {
  21
  22 // static
  23 MojoResult DataPipe::ValidateCreateOptions(
  24     const MojoCreateDataPipeOptions* in_options,
  25     MojoCreateDataPipeOptions* out_options) {
  26   const MojoCreateDataPipeOptionsFlags kKnownFlags =
  27       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD;
  28   static const MojoCreateDataPipeOptions kDefaultOptions = {
  29     static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions)),
  30     MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,
  31     1u,
  32     static_cast<uint32_t>(kDefaultDataPipeCapacityBytes)
  33   };
  34
  35   *out_options = kDefaultOptions;
  36   if (!in_options)
  37     return MOJO_RESULT_OK;
  38
  39   MojoResult result =
  40       ValidateOptionsStructPointerSizeAndFlags<MojoCreateDataPipeOptions>(
  41           in_options, kKnownFlags, out_options);
  42   if (result != MOJO_RESULT_OK)
  43     return result;
  44
  45   // Checks for fields beyond |flags|:
  46
  47   if (!HAS_OPTIONS_STRUCT_MEMBER(MojoCreateDataPipeOptions, element_num_bytes,
  48                                  in_options))
  49     return MOJO_RESULT_OK;
  50   if (in_options->element_num_bytes == 0)
  51     return MOJO_RESULT_INVALID_ARGUMENT;
  52   out_options->element_num_bytes = in_options->element_num_bytes;
  53
  54   if (!HAS_OPTIONS_STRUCT_MEMBER(MojoCreateDataPipeOptions, capacity_num_bytes,
  55                                  in_options) ||
  56       in_options->capacity_num_bytes == 0) {
  57     // Round the default capacity down to a multiple of the element size (but at
  58     // least one element).
  59     out_options->capacity_num_bytes = std::max(
  60         static_cast<uint32_t>(kDefaultDataPipeCapacityBytes -
  61             (kDefaultDataPipeCapacityBytes % out_options->element_num_bytes)),
  62         out_options->element_num_bytes);
  63     return MOJO_RESULT_OK;
  64   }
  65   if (in_options->capacity_num_bytes % out_options->element_num_bytes != 0)
  66     return MOJO_RESULT_INVALID_ARGUMENT;
  67   if (in_options->capacity_num_bytes > kMaxDataPipeCapacityBytes)
  68     return MOJO_RESULT_RESOURCE_EXHAUSTED;
  69   out_options->capacity_num_bytes = in_options->capacity_num_bytes;
  70
  71   return MOJO_RESULT_OK;
  72 }
  73
  74 void DataPipe::ProducerCancelAllWaiters() {
  75   base::AutoLock locker(lock_);
  76   DCHECK(has_local_producer_no_lock());
  77   producer_waiter_list_->CancelAllWaiters();
  78 }
  79
  80 void DataPipe::ProducerClose() {
  81   base::AutoLock locker(lock_);
  82   DCHECK(producer_open_);
  83   producer_open_ = false;
  84   DCHECK(has_local_producer_no_lock());
  85   producer_waiter_list_.reset();
  86   // Not a bug, except possibly in "user" code.
  87   DVLOG_IF(2, producer_in_two_phase_write_no_lock())
  88       << "Producer closed with active two-phase write";
  89   producer_two_phase_max_num_bytes_written_ = 0;
  90   ProducerCloseImplNoLock();
  91   AwakeConsumerWaitersForStateChangeNoLock();
  92 }
  93
  94 MojoResult DataPipe::ProducerWriteData(const void* elements,
  95                                        uint32_t* num_bytes,
  96                                        bool all_or_none) {
  97   base::AutoLock locker(lock_);
  98   DCHECK(has_local_producer_no_lock());
  99
 100   if (producer_in_two_phase_write_no_lock())
 101     return MOJO_RESULT_BUSY;
 102   if (!consumer_open_no_lock())
 103     return MOJO_RESULT_FAILED_PRECONDITION;
 104
 105   // Returning "busy" takes priority over "invalid argument".
 106   if (*num_bytes % element_num_bytes_ != 0)
 107     return MOJO_RESULT_INVALID_ARGUMENT;
 108
 109   if (*num_bytes == 0)
 110     return MOJO_RESULT_OK;  // Nothing to do.
 111
 112   MojoWaitFlags old_consumer_satisfied_flags = ConsumerSatisfiedFlagsNoLock();
 113   MojoResult rv = ProducerWriteDataImplNoLock(elements, num_bytes, all_or_none);
 114   if (ConsumerSatisfiedFlagsNoLock() != old_consumer_satisfied_flags)
 115     AwakeConsumerWaitersForStateChangeNoLock();
 116   return rv;
 117 }
 118
 119 MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
 120                                             uint32_t* buffer_num_bytes,
 121                                             bool all_or_none) {
 122   base::AutoLock locker(lock_);
 123   DCHECK(has_local_producer_no_lock());
 124
 125   if (producer_in_two_phase_write_no_lock())
 126     return MOJO_RESULT_BUSY;
 127   if (!consumer_open_no_lock())
 128     return MOJO_RESULT_FAILED_PRECONDITION;
 129
 130   if (all_or_none && *buffer_num_bytes % element_num_bytes_ != 0)
 131     return MOJO_RESULT_INVALID_ARGUMENT;
 132
 133   MojoResult rv = ProducerBeginWriteDataImplNoLock(buffer, buffer_num_bytes,
 134                                                    all_or_none);
 135   if (rv != MOJO_RESULT_OK)
 136     return rv;
 137   // Note: No need to awake producer waiters, even though we're going from
 138   // writable to non-writable (since you can't wait on non-writability).
 139   // Similarly, though this may have discarded data (in "may discard" mode),
 140   // making it non-readable, there's still no need to awake consumer waiters.
 141   DCHECK(producer_in_two_phase_write_no_lock());
 142   return MOJO_RESULT_OK;
 143 }
 144
 145 MojoResult DataPipe::ProducerEndWriteData(uint32_t num_bytes_written) {
 146   base::AutoLock locker(lock_);
 147   DCHECK(has_local_producer_no_lock());
 148
 149   if (!producer_in_two_phase_write_no_lock())
 150     return MOJO_RESULT_FAILED_PRECONDITION;
 151   // Note: Allow successful completion of the two-phase write even if the
 152   // consumer has been closed.
 153
 154   MojoWaitFlags old_consumer_satisfied_flags = ConsumerSatisfiedFlagsNoLock();
 155   MojoResult rv;
 156   if (num_bytes_written > producer_two_phase_max_num_bytes_written_ ||
 157       num_bytes_written % element_num_bytes_ != 0) {
 158     rv = MOJO_RESULT_INVALID_ARGUMENT;
 159     producer_two_phase_max_num_bytes_written_ = 0;
 160   } else {
 161     rv = ProducerEndWriteDataImplNoLock(num_bytes_written);
 162   }
 163   // Two-phase write ended even on failure.
 164   DCHECK(!producer_in_two_phase_write_no_lock());
 165   // If we're now writable, we *became* writable (since we weren't writable
 166   // during the two-phase write), so awake producer waiters.
 167   if ((ProducerSatisfiedFlagsNoLock() & MOJO_WAIT_FLAG_WRITABLE))
 168     AwakeProducerWaitersForStateChangeNoLock();
 169   if (ConsumerSatisfiedFlagsNoLock() != old_consumer_satisfied_flags)
 170     AwakeConsumerWaitersForStateChangeNoLock();
 171   return rv;
 172 }
 173
 174 MojoResult DataPipe::ProducerAddWaiter(Waiter* waiter,
 175                                        MojoWaitFlags flags,
 176                                        MojoResult wake_result) {
 177   base::AutoLock locker(lock_);
 178   DCHECK(has_local_producer_no_lock());
 179
 180   if ((flags & ProducerSatisfiedFlagsNoLock()))
 181     return MOJO_RESULT_ALREADY_EXISTS;
 182   if (!(flags & ProducerSatisfiableFlagsNoLock()))
 183     return MOJO_RESULT_FAILED_PRECONDITION;
 184
 185   producer_waiter_list_->AddWaiter(waiter, flags, wake_result);
 186   return MOJO_RESULT_OK;
 187 }
 188
 189 void DataPipe::ProducerRemoveWaiter(Waiter* waiter) {
 190   base::AutoLock locker(lock_);
 191   DCHECK(has_local_producer_no_lock());
 192   producer_waiter_list_->RemoveWaiter(waiter);
 193 }
 194
 195 bool DataPipe::ProducerIsBusy() const {
 196   base::AutoLock locker(lock_);
 197   return producer_in_two_phase_write_no_lock();
 198 }
 199
 200 void DataPipe::ConsumerCancelAllWaiters() {
 201   base::AutoLock locker(lock_);
 202   DCHECK(has_local_consumer_no_lock());
 203   consumer_waiter_list_->CancelAllWaiters();
 204 }
 205
 206 void DataPipe::ConsumerClose() {
 207   base::AutoLock locker(lock_);
 208   DCHECK(consumer_open_);
 209   consumer_open_ = false;
 210   DCHECK(has_local_consumer_no_lock());
 211   consumer_waiter_list_.reset();
 212   // Not a bug, except possibly in "user" code.
 213   DVLOG_IF(2, consumer_in_two_phase_read_no_lock())
 214       << "Consumer closed with active two-phase read";
 215   consumer_two_phase_max_num_bytes_read_ = 0;
 216   ConsumerCloseImplNoLock();
 217   AwakeProducerWaitersForStateChangeNoLock();
 218 }
 219
 220 MojoResult DataPipe::ConsumerReadData(void* elements,
 221                                       uint32_t* num_bytes,
 222                                       bool all_or_none) {
 223   base::AutoLock locker(lock_);
 224   DCHECK(has_local_consumer_no_lock());
 225
 226   if (consumer_in_two_phase_read_no_lock())
 227     return MOJO_RESULT_BUSY;
 228
 229   if (*num_bytes % element_num_bytes_ != 0)
 230     return MOJO_RESULT_INVALID_ARGUMENT;
 231
 232   if (*num_bytes == 0)
 233     return MOJO_RESULT_OK;  // Nothing to do.
 234
 235   MojoWaitFlags old_producer_satisfied_flags = ProducerSatisfiedFlagsNoLock();
 236   MojoResult rv = ConsumerReadDataImplNoLock(elements, num_bytes, all_or_none);
 237   if (ProducerSatisfiedFlagsNoLock() != old_producer_satisfied_flags)
 238     AwakeProducerWaitersForStateChangeNoLock();
 239   return rv;
 240 }
 241
 242 MojoResult DataPipe::ConsumerDiscardData(uint32_t* num_bytes,
 243                                          bool all_or_none) {
 244   base::AutoLock locker(lock_);
 245   DCHECK(has_local_consumer_no_lock());
 246
 247   if (consumer_in_two_phase_read_no_lock())
 248     return MOJO_RESULT_BUSY;
 249
 250   if (*num_bytes % element_num_bytes_ != 0)
 251     return MOJO_RESULT_INVALID_ARGUMENT;
 252
 253   if (*num_bytes == 0)
 254     return MOJO_RESULT_OK;  // Nothing to do.
 255
 256   MojoWaitFlags old_producer_satisfied_flags = ProducerSatisfiedFlagsNoLock();
 257   MojoResult rv = ConsumerDiscardDataImplNoLock(num_bytes, all_or_none);
 258   if (ProducerSatisfiedFlagsNoLock() != old_producer_satisfied_flags)
 259     AwakeProducerWaitersForStateChangeNoLock();
 260   return rv;
 261 }
 262
 263 MojoResult DataPipe::ConsumerQueryData(uint32_t* num_bytes) {
 264   base::AutoLock locker(lock_);
 265   DCHECK(has_local_consumer_no_lock());
 266
 267   if (consumer_in_two_phase_read_no_lock())
 268     return MOJO_RESULT_BUSY;
 269
 270   // Note: Don't need to validate |*num_bytes| for query.
 271   return ConsumerQueryDataImplNoLock(num_bytes);
 272 }
 273
 274 MojoResult DataPipe::ConsumerBeginReadData(const void** buffer,
 275                                            uint32_t* buffer_num_bytes,
 276                                            bool all_or_none) {
 277   base::AutoLock locker(lock_);
 278   DCHECK(has_local_consumer_no_lock());
 279
 280   if (consumer_in_two_phase_read_no_lock())
 281     return MOJO_RESULT_BUSY;
 282
 283   if (all_or_none && *buffer_num_bytes % element_num_bytes_ != 0)
 284     return MOJO_RESULT_INVALID_ARGUMENT;
 285
 286   MojoResult rv = ConsumerBeginReadDataImplNoLock(buffer, buffer_num_bytes,
 287                                                   all_or_none);
 288   if (rv != MOJO_RESULT_OK)
 289     return rv;
 290   DCHECK(consumer_in_two_phase_read_no_lock());
 291   return MOJO_RESULT_OK;
 292 }
 293
 294 MojoResult DataPipe::ConsumerEndReadData(uint32_t num_bytes_read) {
 295   base::AutoLock locker(lock_);
 296   DCHECK(has_local_consumer_no_lock());
 297
 298   if (!consumer_in_two_phase_read_no_lock())
 299     return MOJO_RESULT_FAILED_PRECONDITION;
 300
 301   MojoWaitFlags old_producer_satisfied_flags = ProducerSatisfiedFlagsNoLock();
 302   MojoResult rv;
 303   if (num_bytes_read > consumer_two_phase_max_num_bytes_read_ ||
 304       num_bytes_read % element_num_bytes_ != 0) {
 305     rv = MOJO_RESULT_INVALID_ARGUMENT;
 306     consumer_two_phase_max_num_bytes_read_ = 0;
 307   } else {
 308     rv = ConsumerEndReadDataImplNoLock(num_bytes_read);
 309   }
 310   // Two-phase read ended even on failure.
 311   DCHECK(!consumer_in_two_phase_read_no_lock());
 312   // If we're now readable, we *became* readable (since we weren't readable
 313   // during the two-phase read), so awake consumer waiters.
 314   if ((ConsumerSatisfiedFlagsNoLock() & MOJO_WAIT_FLAG_READABLE))
 315     AwakeConsumerWaitersForStateChangeNoLock();
 316   if (ProducerSatisfiedFlagsNoLock() != old_producer_satisfied_flags)
 317     AwakeProducerWaitersForStateChangeNoLock();
 318   return rv;
 319 }
 320
 321 MojoResult DataPipe::ConsumerAddWaiter(Waiter* waiter,
 322                                        MojoWaitFlags flags,
 323                                        MojoResult wake_result) {
 324   base::AutoLock locker(lock_);
 325   DCHECK(has_local_consumer_no_lock());
 326
 327   if ((flags & ConsumerSatisfiedFlagsNoLock()))
 328     return MOJO_RESULT_ALREADY_EXISTS;
 329   if (!(flags & ConsumerSatisfiableFlagsNoLock()))
 330     return MOJO_RESULT_FAILED_PRECONDITION;
 331
 332   consumer_waiter_list_->AddWaiter(waiter, flags, wake_result);
 333   return MOJO_RESULT_OK;
 334 }
 335
 336 void DataPipe::ConsumerRemoveWaiter(Waiter* waiter) {
 337   base::AutoLock locker(lock_);
 338   DCHECK(has_local_consumer_no_lock());
 339   consumer_waiter_list_->RemoveWaiter(waiter);
 340 }
 341
 342 bool DataPipe::ConsumerIsBusy() const {
 343   base::AutoLock locker(lock_);
 344   return consumer_in_two_phase_read_no_lock();
 345 }
 346
 347
 348 DataPipe::DataPipe(bool has_local_producer,
 349                    bool has_local_consumer,
 350                    const MojoCreateDataPipeOptions& validated_options)
 351     : may_discard_((validated_options.flags &
 352                        MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD)),
 353       element_num_bytes_(validated_options.element_num_bytes),
 354       capacity_num_bytes_(validated_options.capacity_num_bytes),
 355       producer_open_(true),
 356       consumer_open_(true),
 357       producer_waiter_list_(has_local_producer ? new WaiterList() : NULL),
 358       consumer_waiter_list_(has_local_consumer ? new WaiterList() : NULL),
 359       producer_two_phase_max_num_bytes_written_(0),
 360       consumer_two_phase_max_num_bytes_read_(0) {
 361   // Check that the passed in options actually are validated.
 362   MojoCreateDataPipeOptions unused ALLOW_UNUSED = { 0 };
 363   DCHECK_EQ(ValidateCreateOptions(&validated_options, &unused), MOJO_RESULT_OK);
 364 }
 365
 366 DataPipe::~DataPipe() {
 367   DCHECK(!producer_open_);
 368   DCHECK(!consumer_open_);
 369   DCHECK(!producer_waiter_list_);
 370   DCHECK(!consumer_waiter_list_);
 371 }
 372
 373 void DataPipe::AwakeProducerWaitersForStateChangeNoLock() {
 374   lock_.AssertAcquired();
 375   if (!has_local_producer_no_lock())
 376     return;
 377   producer_waiter_list_->AwakeWaitersForStateChange(
 378       ProducerSatisfiedFlagsNoLock(), ProducerSatisfiableFlagsNoLock());
 379 }
 380
 381 void DataPipe::AwakeConsumerWaitersForStateChangeNoLock() {
 382   lock_.AssertAcquired();
 383   if (!has_local_consumer_no_lock())
 384     return;
 385   consumer_waiter_list_->AwakeWaitersForStateChange(
 386       ConsumerSatisfiedFlagsNoLock(), ConsumerSatisfiableFlagsNoLock());
 387 }
 388
 389 }  // namespace system
 390 }  // namespace mojo