ptlrpc_request_set_replen(req);
- if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
- req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
- SEQ_CONTROLLER_PORTAL : SEQ_METADATA_PORTAL;
- /* update mdt field of *in, it is required for fld update
- * on super sequence allocator node. */
- if (opc == SEQ_ALLOC_SUPER)
- in->lsr_mdt = seq->lcs_space.lsr_mdt;
+ if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
+ req->rq_request_portal = SEQ_METADATA_PORTAL;
+ in->lsr_flags = LU_SEQ_RANGE_MDT;
} else {
- LASSERT(opc == SEQ_ALLOC_META);
+ LASSERTF(seq->lcs_type == LUSTRE_SEQ_DATA,
+ "unknown lcs_type %u\n", seq->lcs_type);
req->rq_request_portal = SEQ_DATA_PORTAL;
+ in->lsr_flags = LU_SEQ_RANGE_OST;
}
+
+ if (opc == SEQ_ALLOC_SUPER) {
+ /* Update index field of *in, it is required for
+ * FLD update on super sequence allocator node. */
+ in->lsr_index = seq->lcs_space.lsr_index;
+ req->rq_request_portal = SEQ_CONTROLLER_PORTAL;
+ } else {
+ LASSERTF(opc == SEQ_ALLOC_META,
+ "unknown opcode %u\n, opc", opc);
+ }
+
ptlrpc_at_set_req_timeout(req);
mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
}
/* Request sequence-controller node to allocate new meta-sequence. */
-static int seq_client_alloc_meta(struct lu_client_seq *seq,
- const struct lu_env *env)
+static int seq_client_alloc_meta(const struct lu_env *env,
+ struct lu_client_seq *seq)
{
int rc;
ENTRY;
}
/* Allocate new sequence for client. */
-static int seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
+static int seq_client_alloc_seq(const struct lu_env *env,
+ struct lu_client_seq *seq, seqno_t *seqnr)
{
int rc;
ENTRY;
LASSERT(range_is_sane(&seq->lcs_space));
if (range_is_exhausted(&seq->lcs_space)) {
- rc = seq_client_alloc_meta(seq, NULL);
+ rc = seq_client_alloc_meta(env, seq);
if (rc) {
- CERROR("%s: Can't allocate new meta-sequence, "
+ CERROR("%s: Can't allocate new meta-sequence,"
"rc %d\n", seq->lcs_name, rc);
RETURN(rc);
} else {
RETURN(rc);
}
+static int seq_fid_alloc_prep(struct lu_client_seq *seq,
+ cfs_waitlink_t *link)
+{
+ if (seq->lcs_update) {
+ cfs_waitq_add(&seq->lcs_waitq, link);
+ cfs_set_current_state(CFS_TASK_UNINT);
+ cfs_up(&seq->lcs_sem);
+
+ cfs_waitq_wait(link, CFS_TASK_UNINT);
+
+ cfs_down(&seq->lcs_sem);
+ cfs_waitq_del(&seq->lcs_waitq, link);
+ cfs_set_current_state(CFS_TASK_RUNNING);
+ return -EAGAIN;
+ }
+ ++seq->lcs_update;
+ cfs_up(&seq->lcs_sem);
+ return 0;
+}
+
+static void seq_fid_alloc_fini(struct lu_client_seq *seq)
+{
+ LASSERT(seq->lcs_update == 1);
+ cfs_down(&seq->lcs_sem);
+ --seq->lcs_update;
+ cfs_waitq_signal(&seq->lcs_waitq);
+}
+
+/* Allocate the whole seq to the caller*/
+int seq_client_get_seq(const struct lu_env *env,
+ struct lu_client_seq *seq, seqno_t *seqnr)
+{
+ cfs_waitlink_t link;
+ int rc;
+
+ LASSERT(seqnr != NULL);
+ cfs_down(&seq->lcs_sem);
+ cfs_waitlink_init(&link);
+
+ while (1) {
+ rc = seq_fid_alloc_prep(seq, &link);
+ if (rc == 0)
+ break;
+ }
+
+ rc = seq_client_alloc_seq(env, seq, seqnr);
+ if (rc) {
+ CERROR("%s: Can't allocate new sequence, "
+ "rc %d\n", seq->lcs_name, rc);
+ seq_fid_alloc_fini(seq);
+ cfs_up(&seq->lcs_sem);
+ return rc;
+ }
+
+ CDEBUG(D_INFO, "%s: allocate sequence "
+ "[0x%16.16"LPF64"x]\n", seq->lcs_name, *seqnr);
+
+ /*Since the caller require the whole seq,
+ *so marked this seq to be used*/
+ seq->lcs_fid.f_oid = LUSTRE_SEQ_MAX_WIDTH;
+ seq->lcs_fid.f_seq = *seqnr;
+ seq->lcs_fid.f_ver = 0;
+
+ /*
+ * Inform caller that sequence switch is performed to allow it
+ * to setup FLD for it.
+ */
+ seq_fid_alloc_fini(seq);
+ cfs_up(&seq->lcs_sem);
+
+ return rc;
+}
+EXPORT_SYMBOL(seq_client_get_seq);
+
/* Allocate new fid on passed client @seq and save it to @fid. */
-int seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
+int seq_client_alloc_fid(const struct lu_env *env,
+ struct lu_client_seq *seq, struct lu_fid *fid)
{
+ cfs_waitlink_t link;
int rc;
ENTRY;
LASSERT(seq != NULL);
LASSERT(fid != NULL);
+ cfs_waitlink_init(&link);
cfs_down(&seq->lcs_sem);
- if (fid_is_zero(&seq->lcs_fid) ||
- fid_oid(&seq->lcs_fid) >= seq->lcs_width)
- {
+ while (1) {
seqno_t seqnr;
- rc = seq_client_alloc_seq(seq, &seqnr);
+ if (!fid_is_zero(&seq->lcs_fid) &&
+ fid_oid(&seq->lcs_fid) < seq->lcs_width) {
+ /* Just bump last allocated fid and return to caller. */
+ seq->lcs_fid.f_oid += 1;
+ rc = 0;
+ break;
+ }
+
+ rc = seq_fid_alloc_prep(seq, &link);
+ if (rc)
+ continue;
+
+ rc = seq_client_alloc_seq(env, seq, &seqnr);
if (rc) {
CERROR("%s: Can't allocate new sequence, "
"rc %d\n", seq->lcs_name, rc);
+ seq_fid_alloc_fini(seq);
cfs_up(&seq->lcs_sem);
RETURN(rc);
}
* to setup FLD for it.
*/
rc = 1;
- } else {
- /* Just bump last allocated fid and return to caller. */
- seq->lcs_fid.f_oid += 1;
- rc = 0;
+
+ seq_fid_alloc_fini(seq);
+ break;
}
*fid = seq->lcs_fid;
*/
void seq_client_flush(struct lu_client_seq *seq)
{
+ cfs_waitlink_t link;
+
LASSERT(seq != NULL);
+ cfs_waitlink_init(&link);
cfs_down(&seq->lcs_sem);
+
+ while (seq->lcs_update) {
+ cfs_waitq_add(&seq->lcs_waitq, &link);
+ cfs_set_current_state(CFS_TASK_UNINT);
+ cfs_up(&seq->lcs_sem);
+
+ cfs_waitq_wait(&link, CFS_TASK_UNINT);
+
+ cfs_down(&seq->lcs_sem);
+ cfs_waitq_del(&seq->lcs_waitq, &link);
+ cfs_set_current_state(CFS_TASK_RUNNING);
+ }
+
fid_zero(&seq->lcs_fid);
/**
* this id shld not be used for seq range allocation.
* set to -1 for dgb check.
*/
- seq->lcs_space.lsr_mdt = -1;
+ seq->lcs_space.lsr_index = -1;
range_init(&seq->lcs_space);
cfs_up(&seq->lcs_sem);
seq->lcs_type = type;
cfs_sema_init(&seq->lcs_sem, 1);
seq->lcs_width = LUSTRE_SEQ_MAX_WIDTH;
+ cfs_waitq_init(&seq->lcs_waitq);
/* Make sure that things are clear before work is started. */
seq_client_flush(seq);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff