void OSD::do_recovery(
  PG *pg, epoch_t queued, uint64_t reserved_pushes,
  ThreadPool::TPHandle &handle)
{
  uint64_t started = 0;

  /*
   * When the value of osd_recovery_sleep is set greater than zero, recovery
   * ops are scheduled after osd_recovery_sleep amount of time from the previous
   * recovery event's schedule time. This is done by adding a
   * recovery_requeue_callback event, which re-queues the recovery op using
   * queue_recovery_after_sleep.
   */
  float recovery_sleep = get_osd_recovery_sleep();
  {
    Mutex::Locker l(service.recovery_sleep_lock);
    if (recovery_sleep > 0 && service.recovery_needs_sleep) {
      PGRef pgref(pg);
      auto recovery_requeue_callback = new FunctionContext([this, pgref, queued, reserved_pushes](int r) {
        dout(20) << "do_recovery wake up at "
                 << ceph_clock_now()
	         << ", re-queuing recovery" << dendl;
	Mutex::Locker l(service.recovery_sleep_lock);
        service.recovery_needs_sleep = false;
        service.queue_recovery_after_sleep(pgref.get(), queued, reserved_pushes);
      });

      // This is true for the first recovery op and when the previous recovery op
      // has been scheduled in the past. The next recovery op is scheduled after
      // completing the sleep from now.
      if (service.recovery_schedule_time < ceph_clock_now()) {
        service.recovery_schedule_time = ceph_clock_now();
      }
      service.recovery_schedule_time += recovery_sleep;
      service.recovery_sleep_timer.add_event_at(service.recovery_schedule_time,
	                                        recovery_requeue_callback);
      dout(20) << "Recovery event scheduled at "
               << service.recovery_schedule_time << dendl;
      return;
    }
  }

  {
    {
      Mutex::Locker l(service.recovery_sleep_lock);
      service.recovery_needs_sleep = true;
    }

    if (pg->pg_has_reset_since(queued)) {
      goto out;
    }

    assert(!pg->deleting);
    assert(pg->is_peered() && pg->is_primary());

    assert(pg->recovery_queued);
    pg->recovery_queued = false;

    dout(10) << "do_recovery starting " << reserved_pushes << " " << *pg << dendl;
#ifdef DEBUG_RECOVERY_OIDS
    dout(20) << "  active was " << service.recovery_oids[pg->info.pgid] << dendl;
#endif

    bool more = pg->start_recovery_ops(reserved_pushes, handle, &started);
    dout(10) << "do_recovery started " << started << "/" << reserved_pushes 
	     << " on " << *pg << dendl;

    // If no recovery op is started, don't bother to manipulate the RecoveryCtx
    if (!started && (more || !pg->have_unfound())) {
      goto out;
    }

    PG::RecoveryCtx rctx = create_context();
    rctx.handle = &handle;

    /*
     * if we couldn't start any recovery ops and things are still
     * unfound, see if we can discover more missing object locations.
     * It may be that our initial locations were bad and we errored
     * out while trying to pull.
     */
    if (!more && pg->have_unfound()) {
      pg->discover_all_missing(*rctx.query_map);
      if (rctx.query_map->empty()) {
	string action;
        if (pg->state_test(PG_STATE_BACKFILLING)) {
	  auto evt = PG::CephPeeringEvtRef(new PG::CephPeeringEvt(
	    queued,
	    queued,
	    PG::DeferBackfill(cct->_conf->osd_recovery_retry_interval)));
	  pg->queue_peering_event(evt);
	  action = "in backfill";
        } else if (pg->state_test(PG_STATE_RECOVERING)) {
	  auto evt = PG::CephPeeringEvtRef(new PG::CephPeeringEvt(
	    queued,
	    queued,
	    PG::DeferRecovery(cct->_conf->osd_recovery_retry_interval)));
	  pg->queue_peering_event(evt);
	  action = "in recovery";
	} else {
	  action = "already out of recovery/backfill";
	}
	dout(10) << __func__ << ": no luck, giving up on this pg for now (" << action << ")" << dendl;
      } else {
	dout(10) << __func__ << ": no luck, giving up on this pg for now (queue_recovery)" << dendl;
	pg->queue_recovery();
      }
    }

    pg->write_if_dirty(*rctx.transaction);
    OSDMapRef curmap = pg->get_osdmap();
    dispatch_context(rctx, pg, curmap);
  }

 out:
  assert(started <= reserved_pushes);
  service.release_reserved_pushes(reserved_pushes);
}