Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit fc36d0a4 authored by Jeff Layton's avatar Jeff Layton Committed by Ilya Dryomov
Browse files

libceph: abort already submitted but abortable requests when map or pool goes full 



When a Ceph volume hits capacity, a flag is set in the OSD map to
indicate that, and a new map is sprayed around the cluster. With cephfs
we want it to shut down any abortable requests that are in progress with
an -ENOSPC error as they'd just hang otherwise.

Add a new ceph_osdc_abort_on_full helper function to handle this. It
will first check whether there is an out-of-space condition in the
cluster and then walk the tree and abort any request that has
r_abort_on_full set with a -ENOSPC error. Call this new function
directly whenever we get a new OSD map.

Signed-off-by: default avatarJeff Layton <jlayton@redhat.com>
Reviewed-by: default avatarIlya Dryomov <idryomov@gmail.com>
Signed-off-by: default avatarIlya Dryomov <idryomov@gmail.com>
parent a1f4020a

net/ceph/osd_client.c

+34 −0
Original line number Original line Diff line number Diff line
@@ -1807,6 +1807,39 @@ static void abort_request(struct ceph_osd_request *req, int err) 
	complete_request(req, err);

	complete_request(req, err);

}

}





/*

 * Drop all pending requests that are stalled waiting on a full condition to

 * clear, and complete them with ENOSPC as the return code.

 */

static void ceph_osdc_abort_on_full(struct ceph_osd_client *osdc)

{

	struct rb_node *n;



	dout("enter abort_on_full\n");



	if (!ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) && !have_pool_full(osdc))

		goto out;



	for (n = rb_first(&osdc->osds); n; n = rb_next(n)) {

		struct ceph_osd *osd = rb_entry(n, struct ceph_osd, o_node);

		struct rb_node *m;



		m = rb_first(&osd->o_requests);

		while (m) {

			struct ceph_osd_request *req = rb_entry(m,

					struct ceph_osd_request, r_node);

			m = rb_next(m);



			if (req->r_abort_on_full &&

			    (ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) ||

			     pool_full(osdc, req->r_t.target_oloc.pool)))

				abort_request(req, -ENOSPC);

		}

	}

out:

	dout("return abort_on_full\n");

}



static void check_pool_dne(struct ceph_osd_request *req)

static void check_pool_dne(struct ceph_osd_request *req)

{

{

	struct ceph_osd_client *osdc = req->r_osdc;

	struct ceph_osd_client *osdc = req->r_osdc;
@@ -3265,6 +3298,7 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg) 




	kick_requests(osdc, &need_resend, &need_resend_linger);

	kick_requests(osdc, &need_resend, &need_resend_linger);





	ceph_osdc_abort_on_full(osdc);

	ceph_monc_got_map(&osdc->client->monc, CEPH_SUB_OSDMAP,

	ceph_monc_got_map(&osdc->client->monc, CEPH_SUB_OSDMAP,

			  osdc->osdmap->epoch);

			  osdc->osdmap->epoch);

	up_write(&osdc->lock);

	up_write(&osdc->lock);