refactor(uORB): guard DeviceNode _data with the node lock instead of an atomic

_data (the lazily-allocated topic buffer) was a px4::atomic only to make the
double-checked-locking publish safe against concurrent readers. It is allocated
under the node lock() and written exactly once, so reading it under ATOMIC_ENTER
- which is that same lock() on POSIX - serializes the subscriber side against the
allocation without needing an atomic. On NuttX it is a set-once, word-aligned
pointer. The copy() guard is moved inside ATOMIC_ENTER accordingly. (Suggested in
review.)

platforms/common/uORB/uORBDeviceNode.cpp

@@ -57,7 +57,7 @@ uORB::DeviceNode::DeviceNode(const struct orb_metadata *meta, const uint8_t inst
 
 uORB::DeviceNode::~DeviceNode()
 {
-	free(_data.load());
+	free(_data);
 
 	const char *devname = get_devname();
 
@@ -147,17 +147,20 @@ uORB::DeviceNode::write(cdev::file_t *filp, const char *buffer, size_t buflen)
 	 *
 	 * Note that filp will usually be NULL.
 	 */
-	if (nullptr == _data.load()) {
+	if (nullptr == _data) {
 
 #ifdef __PX4_NUTTX
 
 		if (!up_interrupt_context()) {
 #endif /* __PX4_NUTTX */
 
+			// Allocate the buffer under the node lock. It cannot be allocated from
+			// ATOMIC_ENTER (an interrupts-off critical section on NuttX), so the lock
+			// is what serializes the lazy allocation against concurrent readers.
 			lock();
 
 			/* re-check size */
-			if (nullptr == _data.load()) {
+			if (nullptr == _data) {
 				const size_t data_size = _meta->o_size * _meta->o_queue;
 				uint8_t *data = (uint8_t *) px4_cache_aligned_alloc(data_size);
 
@@ -165,7 +168,7 @@ uORB::DeviceNode::write(cdev::file_t *filp, const char *buffer, size_t buflen)
 					memset(data, 0, data_size);
 				}
 
-				_data.store(data);
+				_data = data;
 			}
 
 			unlock();
@@ -176,7 +179,7 @@ uORB::DeviceNode::write(cdev::file_t *filp, const char *buffer, size_t buflen)
 #endif /* __PX4_NUTTX */
 
 		/* failed or could not allocate */
-		if (nullptr == _data.load()) {
+		if (nullptr == _data) {
 			return -ENOMEM;
 		}
 	}
@@ -191,7 +194,7 @@ uORB::DeviceNode::write(cdev::file_t *filp, const char *buffer, size_t buflen)
 	/* wrap-around happens after ~49 days, assuming a publisher rate of 1 kHz */
 	unsigned generation = _generation.fetch_add(1);
 
-	memcpy(_data.load() + (_meta->o_size * (generation % _meta->o_queue)), buffer, _meta->o_size);
+	memcpy(_data + (_meta->o_size * (generation % _meta->o_queue)), buffer, _meta->o_size);
 
 	// callbacks
 	for (auto item : _callbacks) {
@@ -412,10 +415,10 @@ int16_t uORB::DeviceNode::process_add_subscription()
 	// send the data to the remote entity.
 	uORBCommunicator::IChannel *ch = uORB::Manager::get_instance()->get_uorb_communicator();
 
-	if (_data.load() != nullptr && ch != nullptr) { // _data will not be null if there is a publisher.
+	if (_data != nullptr && ch != nullptr) { // _data will not be null if there is a publisher.
 		// Only send the most recent data to initialize the remote end.
 		if (_data_valid) {
-			ch->send_message(_meta->o_name, _meta->o_size, _data.load() + (_meta->o_size * ((_generation.load() - 1) % _meta->o_queue)));
+			ch->send_message(_meta->o_name, _meta->o_size, _data + (_meta->o_size * ((_generation.load() - 1) % _meta->o_queue)));
 		}
 	}
 

platforms/common/uORB/uORBDeviceNode.hpp

@@ -224,42 +224,49 @@ public:
 	 */
 	bool copy(void *dst, unsigned &generation)
 	{
-		if ((dst != nullptr) && (_data.load() != nullptr)) {
-			if (_meta->o_queue == 1) {
-				ATOMIC_ENTER;
-				memcpy(dst, _data.load(), _meta->o_size);
-				generation = _generation.load();
-				ATOMIC_LEAVE;
-				return true;
-
-			} else {
-				ATOMIC_ENTER;
-				const unsigned current_generation = _generation.load();
-
-				if (current_generation == generation) {
-					/* The subscriber already read the latest message, but nothing new was published yet.
-					* Return the previous message
-					*/
-					--generation;
-				}
-
-				// Compatible with normal and overflow conditions
-				if (!is_in_range(current_generation - _meta->o_queue, generation, current_generation - 1)) {
-					// Reader is too far behind: some messages are lost
-					generation = current_generation - _meta->o_queue;
-				}
-
-				memcpy(dst, _data.load() + (_meta->o_size * (generation % _meta->o_queue)), _meta->o_size);
-				ATOMIC_LEAVE;
-
-				++generation;
-
-				return true;
-			}
+		if (dst == nullptr) {
+			return false;
 		}
 
-		return false;
+		// _data is read (and lazily allocated in write()) under the node lock. On POSIX
+		// ATOMIC_ENTER is that same lock, so reading it here serializes against the
+		// allocation; on NuttX it is a set-once, word-aligned pointer.
+		ATOMIC_ENTER;
 
+		if (_data == nullptr) {
+			ATOMIC_LEAVE;
+			return false;
+		}
+
+		if (_meta->o_queue == 1) {
+			memcpy(dst, _data, _meta->o_size);
+			generation = _generation.load();
+			ATOMIC_LEAVE;
+			return true;
+
+		} else {
+			const unsigned current_generation = _generation.load();
+
+			if (current_generation == generation) {
+				/* The subscriber already read the latest message, but nothing new was published yet.
+				* Return the previous message
+				*/
+				--generation;
+			}
+
+			// Compatible with normal and overflow conditions
+			if (!is_in_range(current_generation - _meta->o_queue, generation, current_generation - 1)) {
+				// Reader is too far behind: some messages are lost
+				generation = current_generation - _meta->o_queue;
+			}
+
+			memcpy(dst, _data + (_meta->o_size * (generation % _meta->o_queue)), _meta->o_size);
+			ATOMIC_LEAVE;
+
+			++generation;
+
+			return true;
+		}
 	}
 
 	// add item to list of work items to schedule on node update
@@ -279,7 +286,7 @@ private:
 
 	const orb_metadata *_meta; /**< object metadata information */
 
-	px4::atomic<uint8_t *> _data{nullptr};   /**< allocated object buffer */
+	uint8_t *_data{nullptr};   /**< allocated object buffer, guarded by the node lock (ATOMIC_ENTER) */
 	bool _data_valid{false}; /**< At least one valid data */
 	px4::atomic<unsigned>  _generation{0};  /**< object generation count */
 	List<uORB::SubscriptionCallback *>	_callbacks;