void OSD::ShardedOpWQ::_process(uint32_t thread_index, heartbeat_handle_d *hb)
{
  uint32_t shard_index = thread_index % num_shards;
  ShardData *sdata = shard_list[shard_index];
  assert(NULL != sdata);

  // peek at spg_t
  sdata->sdata_op_ordering_lock.Lock();
  if (sdata->pqueue->empty()) {
    dout(20) << __func__ << " empty q, waiting" << dendl;
    // optimistically sleep a moment; maybe another work item will come along.
    osd->cct->get_heartbeat_map()->reset_timeout(hb,
      osd->cct->_conf->threadpool_default_timeout, 0);
    sdata->sdata_lock.Lock();
    sdata->sdata_op_ordering_lock.Unlock();
    sdata->sdata_cond.WaitInterval(sdata->sdata_lock,
      utime_t(osd->cct->_conf->threadpool_empty_queue_max_wait, 0));
    sdata->sdata_lock.Unlock();
    sdata->sdata_op_ordering_lock.Lock();
    if (sdata->pqueue->empty()) {
      sdata->sdata_op_ordering_lock.Unlock();
      return;
    }
  }
  pair<spg_t, PGQueueable> item = sdata->pqueue->dequeue();
  if (osd->is_stopping()) {
    sdata->sdata_op_ordering_lock.Unlock();
    return;    // OSD shutdown, discard.
  }
  PGRef pg;
  uint64_t requeue_seq;
  {
    auto& slot = sdata->pg_slots[item.first];
    dout(30) << __func__ << " " << item.first
	     << " to_process " << slot.to_process
	     << " waiting_for_pg=" << (int)slot.waiting_for_pg << dendl;
    slot.to_process.push_back(item.second);
    // note the requeue seq now...
    requeue_seq = slot.requeue_seq;
    if (slot.waiting_for_pg) {
      // save ourselves a bit of effort
      dout(20) << __func__ << " " << item.first << " item " << item.second
	       << " queued, waiting_for_pg" << dendl;
      sdata->sdata_op_ordering_lock.Unlock();
      return;
    }
    pg = slot.pg;
    dout(20) << __func__ << " " << item.first << " item " << item.second
	     << " queued" << dendl;
    ++slot.num_running;
  }
  sdata->sdata_op_ordering_lock.Unlock();

  osd->service.maybe_inject_dispatch_delay();

  // [lookup +] lock pg (if we have it)
  if (!pg) {
    pg = osd->_lookup_lock_pg(item.first);
  } else {
    pg->lock();
  }

  osd->service.maybe_inject_dispatch_delay();

  boost::optional<PGQueueable> qi;

  // we don't use a Mutex::Locker here because of the
  // osd->service.release_reserved_pushes() call below
  sdata->sdata_op_ordering_lock.Lock();

  auto q = sdata->pg_slots.find(item.first);
  assert(q != sdata->pg_slots.end());
  auto& slot = q->second;
  --slot.num_running;

  if (slot.to_process.empty()) {
    // raced with wake_pg_waiters or prune_pg_waiters
    dout(20) << __func__ << " " << item.first << " nothing queued" << dendl;
    if (pg) {
      pg->unlock();
    }
    sdata->sdata_op_ordering_lock.Unlock();
    return;
  }
  if (requeue_seq != slot.requeue_seq) {
    dout(20) << __func__ << " " << item.first
	     << " requeue_seq " << slot.requeue_seq << " > our "
	     << requeue_seq << ", we raced with wake_pg_waiters"
	     << dendl;
    if (pg) {
      pg->unlock();
    }
    sdata->sdata_op_ordering_lock.Unlock();
    return;
  }
  if (pg && !slot.pg && !pg->deleting) {
    dout(20) << __func__ << " " << item.first << " set pg to " << pg << dendl;
    slot.pg = pg;
  }
  dout(30) << __func__ << " " << item.first << " to_process " << slot.to_process
	   << " waiting_for_pg=" << (int)slot.waiting_for_pg << dendl;

  // make sure we're not already waiting for this pg
  if (slot.waiting_for_pg) {
    dout(20) << __func__ << " " << item.first << " item " << item.second
	     << " slot is waiting_for_pg" << dendl;
    if (pg) {
      pg->unlock();
    }
    sdata->sdata_op_ordering_lock.Unlock();
    return;
  }

  // take next item
  qi = slot.to_process.front();
  slot.to_process.pop_front();
  dout(20) << __func__ << " " << item.first << " item " << *qi
	   << " pg " << pg << dendl;

  if (!pg) {
    // should this pg shard exist on this osd in this (or a later) epoch?
    OSDMapRef osdmap = sdata->waiting_for_pg_osdmap;
    if (osdmap->is_up_acting_osd_shard(item.first, osd->whoami)) {
      dout(20) << __func__ << " " << item.first
	       << " no pg, should exist, will wait" << " on " << *qi << dendl;
      slot.to_process.push_front(*qi);
      slot.waiting_for_pg = true;
    } else if (qi->get_map_epoch() > osdmap->get_epoch()) {
      dout(20) << __func__ << " " << item.first << " no pg, item epoch is "
	       << qi->get_map_epoch() << " > " << osdmap->get_epoch()
	       << ", will wait on " << *qi << dendl;
      slot.to_process.push_front(*qi);
      slot.waiting_for_pg = true;
    } else {
      dout(20) << __func__ << " " << item.first << " no pg, shouldn't exist,"
	       << " dropping " << *qi << dendl;
      // share map with client?
      if (boost::optional<OpRequestRef> _op = qi->maybe_get_op()) {
	Session *session = static_cast<Session *>(
	  (*_op)->get_req()->get_connection()->get_priv());
	if (session) {
	  osd->maybe_share_map(session, *_op, sdata->waiting_for_pg_osdmap);
	  session->put();
	}
      }
      unsigned pushes_to_free = qi->get_reserved_pushes();
      if (pushes_to_free > 0) {
	sdata->sdata_op_ordering_lock.Unlock();
	osd->service.release_reserved_pushes(pushes_to_free);
	return;
      }
    }
    sdata->sdata_op_ordering_lock.Unlock();
    return;
  }
  sdata->sdata_op_ordering_lock.Unlock();


  // osd_opwq_process marks the point at which an operation has been dequeued
  // and will begin to be handled by a worker thread.
  {
#ifdef WITH_LTTNG
    osd_reqid_t reqid;
    if (boost::optional<OpRequestRef> _op = qi->maybe_get_op()) {
      reqid = (*_op)->get_reqid();
    }
#endif
    tracepoint(osd, opwq_process_start, reqid.name._type,
        reqid.name._num, reqid.tid, reqid.inc);
  }

  lgeneric_subdout(osd->cct, osd, 30) << "dequeue status: ";
  Formatter *f = Formatter::create("json");
  f->open_object_section("q");
  dump(f);
  f->close_section();
  f->flush(*_dout);
  delete f;
  *_dout << dendl;

  ThreadPool::TPHandle tp_handle(osd->cct, hb, timeout_interval,
				 suicide_interval);
  qi->run(osd, pg, tp_handle);

  {
#ifdef WITH_LTTNG
    osd_reqid_t reqid;
    if (boost::optional<OpRequestRef> _op = qi->maybe_get_op()) {
      reqid = (*_op)->get_reqid();
    }
#endif
    tracepoint(osd, opwq_process_finish, reqid.name._type,
        reqid.name._num, reqid.tid, reqid.inc);
  }

  pg->unlock();
}