code: func (r *Rx) Pull() []byte {
  	if r.head == r.tail+jump {
  		// We've already pulled the whole pipe.
  		return nil
  	}
  
  	header := r.p.readAtomic(r.head)
  	if header&slotFree != 0 {
  		// The next slot is free, we can't pull it yet.
  		return nil
  	}
  
  	payloadSize := header & slotSizeMask
  	newHead := r.head + payloadToSlotSize(payloadSize)
  	headWrap := (r.head & revolutionMask) | uint64(len(r.p.buffer))
  
  	// Check if this is a wrapping slot. If that's the case, it carries no
  	// data, so we just skip it and try again from the first slot.
  	if int64(newHead-headWrap) >= 0 {
  		if int64(newHead-headWrap) > int64(jump) || newHead&offsetMask != 0 {
  			return nil
  		}
  
  		if r.tail == r.head {
  			// If this is the first pull since the last Flush()
  			// call, we flush the state so that the sender can use
  			// this space if it needs to.
  			r.p.writeAtomic(r.head, slotFree|slotToPayloadSize(newHead-r.head))
  			r.tail = newHead
  		}
  
  		r.head = newHead
  		return r.Pull()
  	}
  
  	// Grab the buffer before updating r.head.
  	b := r.p.data(r.head, payloadSize)
  	r.head = newHead
  	return b
  }
docs: // Pull reads the next buffer from the pipe, returning nil if there isn't one
  // currently available.
  //
  // The returned slice is available until Flush() is next called. After that, it
  // must not be touched.