1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
// Copyright 2013 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifndef MOJO_SYSTEM_DATA_PIPE_H_
#define MOJO_SYSTEM_DATA_PIPE_H_
#include "base/basictypes.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
#include "base/synchronization/lock.h"
#include "mojo/public/c/system/core.h"
#include "mojo/system/system_impl_export.h"
namespace mojo {
namespace system {
class Waiter;
class WaiterList;
// |DataPipe| is a base class for secondary objects implementing data pipes,
// similar to |MessagePipe| (see the explanatory comment in core.cc). It is
// typically owned by the dispatcher(s) corresponding to the local endpoints.
// Its subclasses implement the three cases: local producer and consumer, local
// producer and remote consumer, and remote producer and local consumer. This
// class is thread-safe.
class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
public base::RefCountedThreadSafe<DataPipe> {
public:
// Validates and/or sets default options. If non-null, |in_options| must point
// to a struct of at least |in_options->struct_size| bytes. |out_options| must
// point to a (current) |MojoCreateDataPipeOptions| and will be entirely
// overwritten on success (it may be partly overwritten on failure).
static MojoResult ValidateOptions(const MojoCreateDataPipeOptions* in_options,
MojoCreateDataPipeOptions* out_options);
// These are called by the producer dispatcher to implement its methods of
// corresponding names.
void ProducerCancelAllWaiters();
void ProducerClose();
// This does not validate its arguments, except to check that |*num_bytes| is
// a multiple of |element_num_bytes_|.
MojoResult ProducerWriteData(const void* elements,
uint32_t* num_bytes,
bool all_or_none);
// This does not validate its arguments.
MojoResult ProducerBeginWriteData(void** buffer,
uint32_t* buffer_num_bytes,
bool all_or_none);
MojoResult ProducerEndWriteData(uint32_t num_bytes_written);
MojoResult ProducerAddWaiter(Waiter* waiter,
MojoWaitFlags flags,
MojoResult wake_result);
void ProducerRemoveWaiter(Waiter* waiter);
bool ProducerIsBusy() const;
// These are called by the consumer dispatcher to implement its methods of
// corresponding names.
void ConsumerCancelAllWaiters();
void ConsumerClose();
// This does not validate its arguments, except to check that |*num_bytes| is
// a multiple of |element_num_bytes_|.
MojoResult ConsumerReadData(void* elements,
uint32_t* num_bytes,
bool all_or_none);
MojoResult ConsumerDiscardData(uint32_t* num_bytes,
bool all_or_none);
MojoResult ConsumerQueryData(uint32_t* num_bytes);
// This does not validate its arguments.
MojoResult ConsumerBeginReadData(const void** buffer,
uint32_t* buffer_num_bytes,
bool all_or_none);
MojoResult ConsumerEndReadData(uint32_t num_bytes_read);
MojoResult ConsumerAddWaiter(Waiter* waiter,
MojoWaitFlags flags,
MojoResult wake_result);
void ConsumerRemoveWaiter(Waiter* waiter);
bool ConsumerIsBusy() const;
protected:
DataPipe(bool has_local_producer,
bool has_local_consumer,
const MojoCreateDataPipeOptions& validated_options);
friend class base::RefCountedThreadSafe<DataPipe>;
virtual ~DataPipe();
virtual void ProducerCloseImplNoLock() = 0;
// |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
virtual MojoResult ProducerWriteDataImplNoLock(const void* elements,
uint32_t* num_bytes,
bool all_or_none) = 0;
virtual MojoResult ProducerBeginWriteDataImplNoLock(
void** buffer,
uint32_t* buffer_num_bytes,
bool all_or_none) = 0;
virtual MojoResult ProducerEndWriteDataImplNoLock(
uint32_t num_bytes_written) = 0;
// Note: A producer should not be writable during a two-phase write.
virtual MojoWaitFlags ProducerSatisfiedFlagsNoLock() = 0;
virtual MojoWaitFlags ProducerSatisfiableFlagsNoLock() = 0;
virtual void ConsumerCloseImplNoLock() = 0;
// |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
virtual MojoResult ConsumerReadDataImplNoLock(void* elements,
uint32_t* num_bytes,
bool all_or_none) = 0;
virtual MojoResult ConsumerDiscardDataImplNoLock(uint32_t* num_bytes,
bool all_or_none) = 0;
// |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
virtual MojoResult ConsumerQueryDataImplNoLock(uint32_t* num_bytes) = 0;
virtual MojoResult ConsumerBeginReadDataImplNoLock(const void** buffer,
uint32_t* buffer_num_bytes,
bool all_or_none) = 0;
virtual MojoResult ConsumerEndReadDataImplNoLock(uint32_t num_bytes_read) = 0;
// Note: A consumer should not be writable during a two-phase read.
virtual MojoWaitFlags ConsumerSatisfiedFlagsNoLock() = 0;
virtual MojoWaitFlags ConsumerSatisfiableFlagsNoLock() = 0;
// Thread-safe and fast (they don't take the lock):
bool may_discard() const { return may_discard_; }
size_t element_num_bytes() const { return element_num_bytes_; }
size_t capacity_num_bytes() const { return capacity_num_bytes_; }
// Must be called under lock.
bool producer_open_no_lock() const {
lock_.AssertAcquired();
return producer_open_;
}
bool consumer_open_no_lock() const {
lock_.AssertAcquired();
return consumer_open_;
}
uint32_t producer_two_phase_max_num_bytes_written_no_lock() const {
lock_.AssertAcquired();
return producer_two_phase_max_num_bytes_written_;
}
uint32_t consumer_two_phase_max_num_bytes_read_no_lock() const {
lock_.AssertAcquired();
return consumer_two_phase_max_num_bytes_read_;
}
void set_producer_two_phase_max_num_bytes_written_no_lock(
uint32_t num_bytes) {
lock_.AssertAcquired();
producer_two_phase_max_num_bytes_written_ = num_bytes;
}
void set_consumer_two_phase_max_num_bytes_read_no_lock(uint32_t num_bytes) {
lock_.AssertAcquired();
consumer_two_phase_max_num_bytes_read_ = num_bytes;
}
bool producer_in_two_phase_write_no_lock() const {
lock_.AssertAcquired();
return producer_two_phase_max_num_bytes_written_ > 0;
}
bool consumer_in_two_phase_read_no_lock() const {
lock_.AssertAcquired();
return consumer_two_phase_max_num_bytes_read_ > 0;
}
private:
void AwakeProducerWaitersForStateChangeNoLock();
void AwakeConsumerWaitersForStateChangeNoLock();
bool has_local_producer_no_lock() const {
lock_.AssertAcquired();
return !!producer_waiter_list_.get();
}
bool has_local_consumer_no_lock() const {
lock_.AssertAcquired();
return !!consumer_waiter_list_.get();
}
const bool may_discard_;
const size_t element_num_bytes_;
const size_t capacity_num_bytes_;
mutable base::Lock lock_; // Protects the following members.
// *Known* state of producer or consumer.
bool producer_open_;
bool consumer_open_;
// Non-null only if the producer or consumer, respectively, is local.
scoped_ptr<WaiterList> producer_waiter_list_;
scoped_ptr<WaiterList> consumer_waiter_list_;
// These are nonzero if and only if a two-phase write/read is in progress.
uint32_t producer_two_phase_max_num_bytes_written_;
uint32_t consumer_two_phase_max_num_bytes_read_;
DISALLOW_COPY_AND_ASSIGN(DataPipe);
};
} // namespace system
} // namespace mojo
#endif // MOJO_SYSTEM_DATA_PIPE_H_
|
