Method OnSubDoneRun

src/brpc/parallel_channel.cpp:215–305

    }
    
    void OnSubDoneRun(SubDone* fin) {
        if (fin != NULL) {
            // [ called from SubDone::Run() ]

            int error_code = fin->cntl.ErrorCode();
            // EPCHANFINISH is not an error of sub calls.
            bool fail = 0 != error_code && EPCHANFINISH != error_code;
            bool cancel =
                // Count failed sub calls, if `fail_limit' is reached, cancel others.
                (fail && _current_fail.fetch_add(1, butil::memory_order_relaxed) + 1
                         == _fail_limit) ||
                // Count successful sub calls, if `success_limit' is reached, cancel others.
                (0 == error_code &&
                 _current_success.fetch_add(1, butil::memory_order_relaxed) + 1
                 == _success_limit);

            if (cancel) {
                // Only cancel once by `fail_limit' or `success_limit'.
                for (int i = 0; i < _ndone; ++i) {
                    SubDone* sd = sub_done(i);
                    if (fin != sd) {
                        bthread_id_error(
                            sd->cntl.call_id(), fail ? ECANCELED : EPCHANFINISH);
                    }
                }
            }
            // NOTE: Don't access any member after the fetch_add because
            // another thread may already go down and Destroy()-ed this object.
            const uint32_t saved_ndone = _ndone;
            const CallId saved_cid = _cntl->_correlation_id;
            // Add 1 to finished sub calls.
            // The release fence is matched with acquire fence below to
            // guarantee visibilities of all other variables.
            const uint32_t val =
                _current_done.fetch_add(1, butil::memory_order_release);
            // Lower 31 bits are number of finished sub calls. If caller is not
            // the last call that finishes, return.
            if ((val & 0x7fffffff) + 1 != saved_ndone) {
                return;
            }
            // If _cntl->call_id() is still there, stop it by sending a special
            // error(which will be cleared) and return.
            if (!(val & 0x80000000)) {
                bthread_id_error(saved_cid, EPCHANFINISH);
                return;
            }
        } else {
            // [ Called from this->Run() ]
            
            // We may cancel sub calls even if all sub calls finish because
            // of reading the value relaxly (and CPU cache is not sync yet).
            // It's OK and we have to, because sub_done can't be accessed
            // after fetch_or.
            uint32_t val = _current_done.load(butil::memory_order_relaxed);
            // Lower 31 bits are number of finished sub calls. Cancel sub calls
            // if not all of them finish.
            if ((val & 0x7fffffff) != (uint32_t)_ndone) {
                for (int i = 0; i < _ndone; ++i) {