fixeria has submitted this change. ( https://gerrit.osmocom.org/c/osmo-trx/+/43116?usp=email )
(
9 is the latest approved patch-set. No files were changed between the latest approved patch-set and the submitted one. )Change subject: Transceiver52M: implement TRXDv2 support ......................................................................
Transceiver52M: implement TRXDv2 support
TRXDv2 was proposed and implemented in osmo-bts a while ago, but osmo-trx was still stuck at TRXDv1. This patch implements TRXDv2 and UL/DL burst batching using libosmo-trx API.
DL: driveTxPriorityQueue() now loops over osmo_trxd_burst_req_parse() to handle datagrams potentially containing multiple batched BURST.req, dispatching each via the new handleBurstReq().
UL: accumulate BURST.ind PDUs for the same FN into one msgb per channel and flush as a single datagram once the next frame's PDU arrives or the channel stops; TRXDv0/v1 channels keep sending one PDU per datagram via sendBurstInd(). Batching is unconditional for now; making it configurable is left as a follow-up.
Change-Id: Ie6a33e3980b3ca4ab05a3d4221d5a8ba0132233d Related: OS#5283 --- M Transceiver52M/Transceiver.cpp M Transceiver52M/Transceiver.h 2 files changed, 173 insertions(+), 56 deletions(-)
Approvals: laforge: Looks good to me, but someone else must approve Jenkins Builder: Verified fixeria: Looks good to me, approved pespin: Looks good to me, but someone else must approve
diff --git a/Transceiver52M/Transceiver.cpp b/Transceiver52M/Transceiver.cpp index 9dc26a4..9405650 100644 --- a/Transceiver52M/Transceiver.cpp +++ b/Transceiver52M/Transceiver.cpp @@ -57,6 +57,13 @@ /* Number of running values use in noise average */ #define NOISE_CNT 20
+/* Worst-case size of a batched TRXDv2 datagram (either direction): up to 8 + * timeslots per frame, each sized as generously as a lone PDU + * (OSMO_TRXD_MSG_BUF_SIZE already covers header + max burst payload for one + * PDU; PDUs after the first omit the 4-byte FN, so this is a safe + * over-estimate, not a tight bound). */ +#define TRX_TRXD_BATCH_BUF_SIZE (8 * OSMO_TRXD_MSG_BUF_SIZE) +
static void dispatch_trx_rate_ctr_change(TransceiverState *state, unsigned int chan) { thread_enable_cancel(false); @@ -146,7 +153,7 @@ mTxPriorityQueueServiceLoopThreads(mChans), mTransmitLatency(wTransmitLatency), mRadioInterface(wRadioInterface), mOn(false),mForceClockInterface(false), mTxFreq(0.0), mRxFreq(0.0), mTSC(0), mMaxExpectedDelayAB(0), mMaxExpectedDelayNB(0), mWriteBurstToDiskMask(0), mVersionTRXD(mChans), mStates(mChans), - mBurstIndMsg(mChans) + mBurstIndMsg(mChans), mBurstIndBatchFn(mChans) { txFullScale = mRadioInterface->fullScaleInputValue(); rxFullScale = mRadioInterface->fullScaleOutputValue(); @@ -156,8 +163,10 @@ mHandover[i][j] = false; }
+ /* Sized for the batched (TRXDv2) case, the larger of the two users of + * this buffer; also plenty for the single-PDU (TRXDv0/v1) case. */ for (size_t i = 0; i < mChans; i++) { - mBurstIndMsg[i] = msgb_alloc(OSMO_TRXD_MSG_BUF_SIZE, "trxd_burst_ind"); + mBurstIndMsg[i] = msgb_alloc(TRX_TRXD_BATCH_BUF_SIZE, "trxd_burst_ind"); OSMO_ASSERT(mBurstIndMsg[i] != NULL); } } @@ -374,6 +383,9 @@ delete mRxServiceLoopThreads[i]; delete mTxPriorityQueueServiceLoopThreads[i];
+ /* Rx thread is joined, so no concurrent access to the batch anymore */ + flushBurstIndBatch(i); + mTxPriorityQueues[i].clear(); }
@@ -1093,29 +1105,9 @@ return 0; }
-bool Transceiver::driveTxPriorityQueue(size_t chan) +bool Transceiver::handleBurstReq(size_t chan, const struct osmo_trxd_burst_req *br) { - char buffer[OSMO_TRXD_MSG_BUF_SIZE]; - int msgLen; - struct osmo_trxd_parse_state st; - struct osmo_trxd_burst_req br; - int rc; - - // check data socket - msgLen = read(mDataSockets[chan], buffer, sizeof(buffer)); - if (msgLen <= 0) { - LOGCHAN(chan, DTRXDDL, NOTICE) << "mDataSockets read(" << mDataSockets[chan] << ") failed: " << msgLen; - return false; - } - - osmo_trxd_parse_state_init(&st); - rc = osmo_trxd_burst_req_parse(&st, &br, (const uint8_t *) buffer, msgLen); - if (rc < 0) { - LOGCHAN(chan, DTRXDDL, ERROR) << "failed to parse BURST.req (rc=" << rc << ")"; - return false; - } - - switch (br.burst_len) { + switch (br->burst_len) { case gSlotLen: /* GSM burst */ break; case EDGE_BURST_NBITS: /* EDGE burst */ @@ -1126,19 +1118,19 @@ break; default: LOGCHAN(chan, DTRXDDL, ERROR) << "badly formatted packet on GSM->TRX interface (burst_len=" - << br.burst_len << ")"; + << br->burst_len << ")"; return false; }
- LOGCHAN(chan, DTRXDDL, DEBUG) << "Rx TRXD message: fn=" << br.fn << ", tn=" << unsigned(br.tn) - << ", burst_len=" << br.burst_len; + LOGCHAN(chan, DTRXDDL, DEBUG) << "Rx TRXD message: fn=" << br->fn << ", tn=" << unsigned(br->tn) + << ", burst_len=" << br->burst_len;
TransceiverState *state = &mStates[chan]; - GSM::Time currTime = GSM::Time(br.fn, br.tn); + GSM::Time currTime = GSM::Time(br->fn, br->tn);
/* Verify proper FN order in DL stream */ - if (state->first_dl_fn_rcv[br.tn]) { - int32_t delta = GSM::FNDelta(currTime.FN(), state->last_dl_time_rcv[br.tn].FN()); + if (state->first_dl_fn_rcv[br->tn]) { + int32_t delta = GSM::FNDelta(currTime.FN(), state->last_dl_time_rcv[br->tn].FN()); if (delta == 1) { /* usual expected scenario, continue code flow */ } else if (delta == 0) { @@ -1148,7 +1140,7 @@ return true; } else if (delta < 0) { LOGCHAN(chan, DTRXDDL, INFO) << "Rx TRXD msg with previous FN " << currTime - << " vs last " << state->last_dl_time_rcv[br.tn]; + << " vs last " << state->last_dl_time_rcv[br->tn]; state->ctrs.tx_trxd_fn_outoforder++; dispatch_trx_rate_ctr_change(state, chan); /* Allow adding radio vector below, since it gets sorted in the queue */ @@ -1158,25 +1150,69 @@ * setups. Also, osmo-trx supports optionally filling empty bursts on * its own. In that case bts-trx is not obliged to submit all bursts. */ LOGCHAN(chan, DTRXDDL, INFO) << "Rx TRXD msg with future FN " << currTime - << " vs last " << state->last_dl_time_rcv[br.tn] + << " vs last " << state->last_dl_time_rcv[br->tn] << ", " << delta - 1 << " FN lost"; state->ctrs.tx_trxd_fn_skipped += delta - 1; dispatch_trx_rate_ctr_change(state, chan); } if (delta > 0) - state->last_dl_time_rcv[br.tn] = currTime; + state->last_dl_time_rcv[br->tn] = currTime; } else { /* Initial check, simply store state */ - state->first_dl_fn_rcv[br.tn] = true; - state->last_dl_time_rcv[br.tn] = currTime; + state->first_dl_fn_rcv[br->tn] = true; + state->last_dl_time_rcv[br->tn] = currTime; }
- BitVector newBurst(br.burst_len); + BitVector newBurst(br->burst_len); BitVector::iterator itr = newBurst.begin(); - const ubit_t *bufferItr = br.burst; + const ubit_t *bufferItr = br->burst; while (itr < newBurst.end()) *itr++ = *bufferItr++;
- addRadioVector(chan, newBurst, br.att, currTime); + addRadioVector(chan, newBurst, br->att, currTime); + + return true; +} + +bool Transceiver::driveTxPriorityQueue(size_t chan) +{ + char buffer[TRX_TRXD_BATCH_BUF_SIZE]; + int msgLen; + struct osmo_trxd_parse_state st; + struct osmo_trxd_burst_req br; + const uint8_t *pos; + size_t remain; + int rc; + + // check data socket + msgLen = read(mDataSockets[chan], buffer, sizeof(buffer)); + if (msgLen <= 0) { + LOGCHAN(chan, DTRXDDL, NOTICE) << "mDataSockets read(" << mDataSockets[chan] << ") failed: " << msgLen; + return false; + } + + /* Normally a single PDU per datagram (TRXDv0/v1, or unbatched TRXDv2), + * but the BTS side may also send us a TRXDv2 batch (multiple PDUs for + * the same FN, one per timeslot): keep parsing while BATCH.ind is set. */ + osmo_trxd_parse_state_init(&st); + pos = (const uint8_t *) buffer; + remain = (size_t) msgLen; + + do { + rc = osmo_trxd_burst_req_parse(&st, &br, pos, remain); + /* rc == 0 shouldn't happen (a parsed PDU is never empty), but treat it + * as an error too: rc <= 0 here would leave pos/remain stuck, spinning + * this loop forever on a re-parse of the exact same input. */ + if (rc <= 0) { + LOGCHAN(chan, DTRXDDL, ERROR) << "failed to parse BURST.req (rc=" << rc << ")"; + return false; + } + + if (!handleBurstReq(chan, &br)) + return false; + + pos += rc; + remain -= rc; + } while (br.flags & OSMO_TRXD_F_BATCH_IND);
return true; } @@ -1226,25 +1262,14 @@ << " bits: " << os; }
-bool Transceiver::driveReceiveFIFO(size_t chan) + +bool Transceiver::sendBurstInd(size_t chan, const struct osmo_trxd_burst_ind *bi) { - struct osmo_trxd_burst_ind bi; struct msgb *msg; int rc;
- if ((rc = pullRadioVector(chan, &bi)) < 0) { - if (rc == -ENOENT) { /* timeslot off, continue processing */ - LOGCHAN(chan, DTRXDUL, DEBUG) << unsigned(bi.tn) << ":" << bi.fn << " timeslot is off"; - return true; - } - return false; /* other errors: we want to stop the process */ - } - - if (!(bi.flags & OSMO_TRXD_F_NOPE_IND) && log_check_level(DTRXDUL, LOGL_DEBUG)) - logRxBurst(chan, &bi); - msg = mBurstIndMsg[chan]; - rc = osmo_trxd_burst_ind_build(msg, mVersionTRXD[chan], &bi); + rc = osmo_trxd_burst_ind_build(msg, mVersionTRXD[chan], bi); if (rc < 0) { LOGCHAN(chan, DTRXDUL, ERROR) << "failed to build BURST.ind (rc=" << rc << ")"; msgb_trim(msg, 0); @@ -1262,6 +1287,74 @@ return true; }
+bool Transceiver::flushBurstIndBatch(size_t chan) +{ + struct msgb *msg = mBurstIndMsg[chan]; + int rc; + + if (msgb_length(msg) == 0) + return true; /* nothing accumulated */ + + osmo_trxd_build_fin(msg, mVersionTRXD[chan]); + + rc = write(mDataSockets[chan], msgb_data(msg), msgb_length(msg)); + msgb_trim(msg, 0); + if (rc < 0) { + LOGCHAN(chan, DTRXDUL, NOTICE) << "mDataSockets write(" << mDataSockets[chan] << ") failed: " << rc; + return false; + } + + return true; +} + +bool Transceiver::queueBurstIndBatched(size_t chan, const struct osmo_trxd_burst_ind *bi) +{ + struct msgb *msg = mBurstIndMsg[chan]; + int rc; + + if (msgb_length(msg) > 0 && mBurstIndBatchFn[chan] != bi->fn) { + /* previous frame's batch is complete (a PDU for a new FN showed up) */ + if (!flushBurstIndBatch(chan)) + return false; + } + + if (msgb_length(msg) == 0) + mBurstIndBatchFn[chan] = bi->fn; + + rc = osmo_trxd_burst_ind_build(msg, mVersionTRXD[chan], bi); + if (rc < 0) { + LOGCHAN(chan, DTRXDUL, ERROR) << "failed to build batched BURST.ind (rc=" << rc << ")"; + msgb_trim(msg, 0); + return false; + } + + return true; +} + +bool Transceiver::driveReceiveFIFO(size_t chan) +{ + struct osmo_trxd_burst_ind bi; + int rc; + + if ((rc = pullRadioVector(chan, &bi)) < 0) { + if (rc == -ENOENT) { /* timeslot off, continue processing */ + LOGCHAN(chan, DTRXDUL, DEBUG) << unsigned(bi.tn) << ":" << bi.fn << " timeslot is off"; + return true; + } + return false; /* other errors: we want to stop the process */ + } + + if (!(bi.flags & OSMO_TRXD_F_NOPE_IND) && log_check_level(DTRXDUL, LOGL_DEBUG)) + logRxBurst(chan, &bi); + + /* TODO: make batching configurable via VTY, so it can be disabled even + * when TRXDv2 is negotiated (e.g. to trade datagram count for latency) */ + if (mVersionTRXD[chan] < 2) + return sendBurstInd(chan, &bi); + + return queueBurstIndBatched(chan, &bi); +} + void Transceiver::driveTxFIFO() {
diff --git a/Transceiver52M/Transceiver.h b/Transceiver52M/Transceiver.h index 53b887e..5c10855 100644 --- a/Transceiver52M/Transceiver.h +++ b/Transceiver52M/Transceiver.h @@ -44,7 +44,7 @@ extern Transceiver *transceiver;
/* The latest TRXD header format version advertised/accepted by this TRX implementation */ -#define TRX_DATA_FORMAT_VER 1 +#define TRX_DATA_FORMAT_VER 2
/** Channel descriptor for transceiver object and channel number pair */ struct TrxChanThParams { @@ -209,6 +209,19 @@ /** Pull and demodulate a burst from the receive FIFO */ int pullRadioVector(size_t chan, struct osmo_trxd_burst_ind *ind);
+ /** Send one BURST.ind PDU as its own datagram (TRXDv0/v1) */ + bool sendBurstInd(size_t chan, const struct osmo_trxd_burst_ind *bi); + + /** Append one BURST.ind PDU to the per-channel TRXDv2 batch, flushing the + * previous frame's batch first if this PDU belongs to a new frame */ + bool queueBurstIndBatched(size_t chan, const struct osmo_trxd_burst_ind *bi); + + /** Write out and reset the accumulated TRXDv2 batch for a channel, if any */ + bool flushBurstIndBatch(size_t chan); + + /** Handle one parsed BURST.req PDU (one out of a batch, or a lone one) */ + bool handleBurstReq(size_t chan, const struct osmo_trxd_burst_req *br); + /** Set modulus for specific timeslot */ void setModulus(size_t timeslot, size_t chan);
@@ -238,10 +251,21 @@ std::vector<unsigned> mVersionTRXD; ///< Format version to use for TRXD protocol communication, per channel std::vector<TransceiverState> mStates;
- /* Per-channel BURST.ind scratch buffer (driveReceiveFIFO()): allocated - * once in the constructor, then reused (trimmed, not freed) for every - * burst to avoid an alloc/free pair on the Rx hot path. */ + /* Per-channel BURST.ind Tx scratch buffer (sendBurstInd(), + * queueBurstIndBatched()/flushBurstIndBatch()): allocated once in the + * constructor, then reused (trimmed, not freed) for every burst to avoid + * an alloc/free pair on the Rx hot path. + * + * Doubles as the TRXDv2 uplink PDU batching accumulator (see + * osmo-gsm-manuals trx_if.adoc, combination "b"): all BURST.ind PDUs for + * one TDMA frame on a given channel are accumulated into a single + * datagram, flushed once the next frame's PDU arrives (i.e. keyed off a + * change in FN, not off tn==7, since not every timeslot is necessarily + * active). Batching across channels (TRXN) is not supported. A channel + * only ever exercises one of the two usages, decided once by the + * negotiated TRXD version (mVersionTRXD), so the two never conflict. */ std::vector<struct msgb *> mBurstIndMsg; + std::vector<uint32_t> mBurstIndBatchFn;
/** Start and stop I/O threads through the control socket API */ bool start();
-
fixeria