};
#define MDS_BFLAG_UNCOMMITTED_WRITES 0x1
+#define MDS_BFLAG_CLOSE_EPOCH 0x2
+#define MDS_BFLAG_DIRTY_EPOCH 0x4
struct mds_body {
struct lustre_id id1;
#define LLI_F_HAVE_OST_SIZE_LOCK 0
#define LLI_F_HAVE_MDS_SIZE_LOCK 1
#define LLI_F_PREFER_EXTENDED_SIZE 2
-
+#define LLI_F_DIRTY_HANDLE 3
+
struct ll_inode_info {
int lli_size_pid;
int lli_inode_magic;
#define LLI_HAVE_FLSIZE(inode) \
test_bit(LLI_F_HAVE_MDS_SIZE_LOCK, &ll_i2info(inode)->lli_flags)
+#define LLI_DIRTY_HANDLE(inode) \
+ test_bit(LLI_F_DIRTY_HANDLE, &ll_i2info(inode)->lli_flags)
/* lprocfs.c */
enum {
//#define SM_DIRTY_WRITE 0x10
#define SM_DO_COW 0x20
#define SM_DO_COWED 0x40
+#define SM_HND_IBLOCKS 0x80
/*
#define SMFS_DO_REC(smfs_info) (smfs_info->smsi_flags & SM_DO_REC)
#define SMFS_DO_INODE_COWED(inode) (I2SMI(inode)->smi_flags & SM_DO_COWED)
#define SMFS_CLEAN_INODE_COWED(inode) (I2SMI(inode)->smi_flags &= ~SM_DO_COWED)
+#define SMFS_DO_HND_IBLOCKS(smfs_info) (smfs_info->smsi_flags & SM_HND_IBLOCKS)
+#define SMFS_SET_HND_IBLOCKS(smfs_info) (smfs_info->smsi_flags |= SM_HND_IBLOCKS)
+#define SMFS_CLEAN_HND_IBLOCKS(smfs_info) (smfs_info->smsi_flags &= ~SM_HND_IBLOCKS)
//#define LVFS_SMFS_BACK_ATTR "lvfs_back_attr"
obd->obd_next_recovery_transno = req_transno;
wake_up = 1;
} else if (queue_len == atomic_read(&obd->obd_req_replay_clients)) {
- /* some clients haven't connected in time, but we need
- * their requests to continue recovery. so, we abort ... */
- CDEBUG(D_ERROR, "abort due to missed clients: queue: %d max: %d\n",
- queue_len, max);
- obd->obd_abort_recovery = 1;
+ /* some clients haven't connected in time, but we can try
+ * to replay requests that demand on already committed ones
+ * also, we can replay first non-committed transation */
+ LASSERT(req_transno != 0);
+ if (req_transno == obd->obd_last_committed + 1) {
+ obd->obd_next_recovery_transno = req_transno;
+ } else if (req_transno > obd->obd_last_committed) {
+ /* can't continue recovery: have no needed transno */
+ obd->obd_abort_recovery = 1;
+ CDEBUG(D_ERROR, "abort due to missed clients. max: %d, "
+ "connected: %d, completed: %d, queue_len: %d, "
+ "req_transno: "LPU64", next_transno: "LPU64"\n",
+ max, connected, completed, queue_len,
+ req_transno, next_transno);
+ }
wake_up = 1;
}
spin_unlock_bh(&obd->obd_processing_task_lock);
/* If some clients haven't connected in time, evict them */
if (obd->obd_abort_recovery) {
int stale;
- CERROR("some clients haven't connect in time (%d/%d),"
+ CDEBUG(D_ERROR, "few clients haven't connect in time (%d/%d),"
"evict them ...\n", obd->obd_connected_clients,
obd->obd_max_recoverable_clients);
obd->obd_abort_recovery = 0;
}
/* next stage: replay requests */
- CWARN("1: request replay stage - %d clients from t"LPU64"\n",
+ CDEBUG(D_ERROR, "1: request replay stage - %d clients from t"LPU64"\n",
atomic_read(&obd->obd_req_replay_clients),
obd->obd_next_recovery_transno);
while ((req = target_next_replay_req(obd))) {
/* If some clients haven't replayed requests in time, evict them */
if (obd->obd_abort_recovery) {
int stale;
- CERROR("req replay timed out, aborting ...\n");
+ CDEBUG(D_ERROR, "req replay timed out, aborting ...\n");
obd->obd_abort_recovery = 0;
stale = class_disconnect_stale_exports(obd, req_replay_done, 0);
atomic_sub(stale, &obd->obd_lock_replay_clients);
abort_req_replay_queue(obd);
+ LBUG();
}
/* The second stage: replay locks */
- CWARN("2: lock replay stage - %d clients\n",
+ CDEBUG(D_ERROR, "2: lock replay stage - %d clients\n",
atomic_read(&obd->obd_lock_replay_clients));
while ((req = target_next_replay_lock(obd))) {
LASSERT(trd->trd_processing_task == current->pid);
}
int ll_md_och_close(struct obd_export *md_exp, struct inode *inode,
- struct obd_client_handle *och)
+ struct obd_client_handle *och, int dirty)
{
struct ptlrpc_request *req = NULL;
struct obdo *obdo = NULL;
RETURN(-ENOMEM);
obdo->o_id = inode->i_ino;
+ obdo->o_generation = inode->i_generation;
obdo->o_valid = OBD_MD_FLID;
obdo_from_inode(obdo, inode, (OBD_MD_FLTYPE | OBD_MD_FLMODE |
OBD_MD_FLATIME | OBD_MD_FLMTIME |
obdo->o_valid |= OBD_MD_FLEPOCH;
obdo->o_easize = ll_i2info(inode)->lli_io_epoch;
- if (ll_validate_size(inode, &obdo->o_size, &obdo->o_blocks))
- obdo->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
+ if (dirty) {
+ /* we modified data through this handle */
+ obdo->o_flags |= MDS_BFLAG_DIRTY_EPOCH;
+ obdo->o_valid |= OBD_MD_FLFLAGS;
+ if (ll_validate_size(inode, &obdo->o_size, &obdo->o_blocks))
+ obdo->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
+ }
rc = md_close(md_exp, obdo, och, &req);
obdo_free(obdo);
struct inode *inode, int flags)
{
struct ll_inode_info *lli = ll_i2info(inode);
+ int freeing = inode->i_state & I_FREEING;
struct obd_client_handle **och_p;
struct obd_client_handle *och;
__u64 *och_usecount;
- int rc = 0;
+ int rc = 0, dirty = 0;
ENTRY;
if (flags & FMODE_WRITE) {
up(&lli->lli_och_sem);
RETURN(0);
}
- och = *och_p;
+ if (ll_is_inode_dirty(inode)) {
+ /* the inode still has dirty pages, let's close later */
+ CDEBUG(D_INODE, "inode %lu/%u still has dirty pages\n",
+ inode->i_ino, inode->i_generation);
+ LASSERT(freeing == 0);
+ ll_queue_done_writing(inode);
+ up(&lli->lli_och_sem);
+ RETURN(0);
+ }
+
+ if (LLI_DIRTY_HANDLE(inode) && (flags & FMODE_WRITE)) {
+ clear_bit(LLI_F_DIRTY_HANDLE, &lli->lli_flags);
+ dirty = 1;
+ } else if (0 && !(flags & FMODE_SYNC) && !freeing) {
+ /* in order to speed up creation rate we pass
+ * closing to dedicated thread so we don't need
+ * to wait for close reply here -bzzz */
+ ll_queue_done_writing(inode);
+ up(&lli->lli_och_sem);
+ RETURN(0);
+ }
+ och = *och_p;
*och_p = NULL;
+
+
up(&lli->lli_och_sem);
/*
* and this will be called from block_ast callack.
*/
if (och && och->och_fh.cookie != DEAD_HANDLE_MAGIC)
- rc = ll_md_och_close(md_exp, inode, och);
+ rc = ll_md_och_close(md_exp, inode, och, dirty);
RETURN(rc);
}
ll_och_fill(inode, it, och);
/* ll_md_och_close() will free och */
- ll_md_och_close(ll_i2mdexp(inode), inode, och);
+ ll_md_och_close(ll_i2mdexp(inode), inode, och, 0);
}
(*och_usecount)++;
CDEBUG(D_INFO, "Writing inode %lu, "LPSZ" bytes, offset %Lu\n",
inode->i_ino, count, *ppos);
+ /* mark open handle dirty */
+ set_bit(LLI_F_DIRTY_HANDLE, &(ll_i2info(inode)->lli_flags));
+
/* generic_file_write handles O_APPEND after getting i_sem */
retval = generic_file_write(file, buf, count, ppos);
EXIT;
rc = ll_file_release(f->f_dentry->d_inode, f);
/* Now also destroy our supplemental och */
- ll_md_och_close(ll_i2mdexp(inode), f->f_dentry->d_inode, och);
+ ll_md_och_close(ll_i2mdexp(inode), f->f_dentry->d_inode, och, 0);
EXIT;
out:
ll_intent_release(&oit);
void llap_write_pending(struct inode *inode, struct ll_async_page *llap)
{
struct ll_inode_info *lli = ll_i2info(inode);
+ struct page *page = llap->llap_page;
spin_lock(&lli->lli_lock);
- list_add(&llap->llap_pending_write, &lli->lli_pending_write_llaps);
+ CDEBUG(D_INODE, "track page 0x%p/%lu %s\n",
+ page, (unsigned long) page->index,
+ !list_empty(&llap->llap_pending_write) ? "(already)" : "");
+ if (list_empty(&llap->llap_pending_write))
+ list_add(&llap->llap_pending_write,
+ &lli->lli_pending_write_llaps);
spin_unlock(&lli->lli_lock);
}
{
struct ll_inode_info *lli = ll_i2info(inode);
struct ll_close_queue *lcq = ll_i2sbi(inode)->ll_lcq;
+ int added = 0;
spin_lock(&lli->lli_lock);
if (lli->lli_send_done_writing &&
list_empty(&lli->lli_pending_write_llaps)) {
-
spin_lock(&lcq->lcq_lock);
if (list_empty(&lli->lli_close_item)) {
CDEBUG(D_INODE, "adding inode %lu/%u to close list\n",
inode->i_ino, inode->i_generation);
- LASSERT(igrab(inode) == inode);
list_add_tail(&lli->lli_close_item, &lcq->lcq_list);
wake_up(&lcq->lcq_waitq);
+ added = 1;
}
spin_unlock(&lcq->lcq_lock);
}
spin_unlock(&lli->lli_lock);
+
+ /*
+ * we can't grab inode under lli_lock, because:
+ * ll_try_done_writing: ll_prep_inode:
+ * spin_lock(&lli_lock) spin_lock(&inode_lock)
+ * igrab() ll_update_inode()
+ * spin_lock(&inode_lock) spin_lock(&lli_lock)
+ */
+ if (added)
+ LASSERT(igrab(inode) == inode);
}
/* The MDS needs us to get the real file attributes, then send a DONE_WRITING */
struct ll_inode_info *lli = ll_i2info(inode);
ENTRY;
+ CDEBUG(D_INODE, "queue closing for %lu/%u\n",
+ inode->i_ino, inode->i_generation);
spin_lock(&lli->lli_lock);
lli->lli_send_done_writing = 1;
spin_unlock(&lli->lli_lock);
EXIT;
}
-#if 0
/* If we know the file size and have the cookies:
* - send a DONE_WRITING rpc
*
* - get the authoritative size and all cookies with GETATTRs
* - send a DONE_WRITING rpc
*/
-static void ll_close_done_writing(struct inode *inode)
+static void ll_try_to_close(struct inode *inode)
{
- struct ll_inode_info *lli = ll_i2info(inode);
- ldlm_policy_data_t policy = { .l_extent = {0, OBD_OBJECT_EOF } };
- struct lustre_handle lockh = { 0 };
- struct obdo *obdo = NULL;
- int rc, ast_flags = 0;
- obd_valid valid;
- ENTRY;
-
- obdo = obdo_alloc();
- if (obdo == NULL) {
- CERROR("cannot allocate obdo, error %d\n",
- -ENOMEM);
- EXIT;
- return;
- }
-
- if (test_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags))
- goto rpc;
-
- rc = ll_extent_lock(NULL, inode, lli->lli_smd, LCK_PW, &policy, &lockh,
- ast_flags, &ll_i2sbi(inode)->ll_done_stime);
- if (rc != 0) {
- CERROR("lock acquisition failed (%d): unable to send "
- "DONE_WRITING for inode %lu/%u\n", rc, inode->i_ino,
- inode->i_generation);
- GOTO(out, rc);
- }
-
- rc = ll_lsm_getattr(ll_i2dtexp(inode), lli->lli_smd, obdo);
- if (rc) {
- CERROR("inode_getattr failed (%d): unable to send DONE_WRITING "
- "for inode %lu/%u\n", rc, inode->i_ino,
- inode->i_generation);
- ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
- GOTO(out, rc);
- }
-
- obdo_refresh_inode(inode, obdo, valid);
-
- CDEBUG(D_INODE, "objid "LPX64" size %Lu, blocks %lu, blksize %lu\n",
- lli->lli_smd->lsm_object_id, inode->i_size, inode->i_blocks,
- inode->i_blksize);
-
- set_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags);
-
- rc = ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
- if (rc != ELDLM_OK)
- CERROR("unlock failed (%d)? proceeding anyways...\n", rc);
-
-rpc:
- obdo->o_id = inode->i_ino;
- obdo->o_size = inode->i_size;
- obdo->o_blocks = inode->i_blocks;
- obdo->o_valid = OBD_MD_FLID | OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
-
- rc = md_done_writing(ll_i2sbi(inode)->ll_mdc_exp, obdo);
-out:
- obdo_free(obdo);
+ struct ll_sb_info *sbi = ll_i2sbi(inode);
+ ll_md_real_close(sbi->ll_md_exp, inode, FMODE_WRITE | FMODE_SYNC);
}
-#endif
static struct ll_inode_info *ll_close_next_lli(struct ll_close_queue *lcq)
{
else if (!list_empty(&lcq->lcq_list)) {
lli = list_entry(lcq->lcq_list.next, struct ll_inode_info,
lli_close_item);
- list_del(&lli->lli_close_item);
+ list_del_init(&lli->lli_close_item);
}
spin_unlock(&lcq->lcq_lock);
while (1) {
struct l_wait_info lwi = { 0 };
struct ll_inode_info *lli;
- //struct inode *inode;
+ struct inode *inode;
l_wait_event_exclusive(lcq->lcq_waitq,
(lli = ll_close_next_lli(lcq)) != NULL,
if (IS_ERR(lli))
break;
- //inode = ll_info2i(lli);
- //ll_close_done_writing(inode);
- //iput(inode);
+ inode = ll_info2i(lli);
+ ll_try_to_close(inode);
+ iput(inode);
}
complete(&lcq->lcq_comp);
int ll_md_close(struct obd_export *md_exp, struct inode *inode,
struct file *file);
int ll_md_och_close(struct obd_export *md_exp, struct inode *inode,
- struct obd_client_handle *och);
+ struct obd_client_handle *och, int dirty);
void ll_och_fill(struct inode *inode, struct lookup_intent *it,
struct obd_client_handle *och);
data->mod_time = LTIME_S(CURRENT_TIME);
}
+/* pass this flag to ll_md_real_close() to send close rpc right away */
+#define FMODE_SYNC 00000010
+
+
#endif /* LLITE_INTERNAL_H */
lli->lli_maxbytes = PAGE_CACHE_MAXBYTES;
spin_lock_init(&lli->lli_lock);
INIT_LIST_HEAD(&lli->lli_pending_write_llaps);
+ INIT_LIST_HEAD(&lli->lli_close_item);
lli->lli_inode_magic = LLI_INODE_MAGIC;
memset(&lli->lli_id, 0, sizeof(lli->lli_id));
sema_init(&lli->lli_och_sem, 1);
#include <linux/iobuf.h>
#endif
+#include <linux/pagevec.h>
+
#define DEBUG_SUBSYSTEM S_LLITE
#include <linux/lustre_mds.h>
RETURN(rc);
}
+
+static void ll_close_vma(struct vm_area_struct *vma)
+{
+ struct inode *inode = vma->vm_file->f_dentry->d_inode;
+ struct address_space *mapping = inode->i_mapping;
+ unsigned long next, size, end;
+ struct ll_async_page *llap;
+ struct obd_export *exp;
+ struct pagevec pvec;
+ int i;
+
+ if (!(vma->vm_flags & VM_SHARED))
+ return;
+
+ /* all pte's are synced to mem_map by the moment
+ * we scan backing store and put all dirty pages
+ * onto pending list to track flushing */
+
+ LASSERT(LLI_DIRTY_HANDLE(inode));
+ exp = ll_i2dtexp(inode);
+ if (exp == NULL) {
+ CERROR("can't get export for the inode\n");
+ return;
+ }
+
+ pagevec_init(&pvec, 0);
+ next = vma->vm_pgoff;
+ size = (vma->vm_end - vma->vm_start) / PAGE_SIZE;
+ end = next + size - 1;
+
+ CDEBUG(D_INODE, "close vma 0x%p[%lu/%lu/%lu from %lu/%u]\n", vma,
+ next, size, end, inode->i_ino, inode->i_generation);
+
+ while (next <= end && pagevec_lookup(&pvec, mapping, next, PAGEVEC_SIZE)) {
+ for (i = 0; i < pagevec_count(&pvec); i++) {
+ struct page *page = pvec.pages[i];
+
+ if (page->index > next)
+ next = page->index;
+ if (next > end)
+ continue;
+ next++;
+
+ lock_page(page);
+ if (page->mapping != mapping || !PageDirty(page)) {
+ unlock_page(page);
+ continue;
+ }
+
+ llap = llap_from_page(page, LLAP_ORIGIN_COMMIT_WRITE);
+ if (IS_ERR(llap)) {
+ CERROR("can't get llap\n");
+ unlock_page(page);
+ continue;
+ }
+
+ llap_write_pending(inode, llap);
+ unlock_page(page);
+ }
+ pagevec_release(&pvec);
+ }
+}
+
static struct vm_operations_struct ll_file_vm_ops = {
.nopage = ll_nopage,
+ .close = ll_close_vma,
};
int ll_file_mmap(struct file * file, struct vm_area_struct * vma)
ENTRY;
rc = generic_file_mmap(file, vma);
- if (rc == 0)
+ if (rc == 0) {
+ struct ll_inode_info *lli = ll_i2info(file->f_dentry->d_inode);
vma->vm_ops = &ll_file_vm_ops;
+ /* mark i/o epoch dirty */
+ if (vma->vm_flags & VM_SHARED)
+ set_bit(LLI_F_DIRTY_HANDLE, &lli->lli_flags);
+ }
RETURN(rc);
}
int rc = 0;
ENTRY;
+ LASSERT(LLI_DIRTY_HANDLE(inode));
LASSERT(PageLocked(page));
(void)llap_cast_private(page); /* assertion */
if (llap == NULL)
RETURN(ERR_PTR(-ENOMEM));
llap->llap_magic = LLAP_MAGIC;
+ INIT_LIST_HEAD(&llap->llap_pending_write);
rc = obd_prep_async_page(exp, ll_i2info(inode)->lli_smd, NULL, page,
(obd_off)page->index << PAGE_SHIFT,
&ll_async_page_ops, llap, &llap->llap_cookie);
OBD_BRW_WRITE, 0, 0, 0, async_flags);
if (rc == 0) {
LL_CDEBUG_PAGE(D_PAGE, llap->llap_page, "write queued\n");
- //llap_write_pending(inode, llap);
+ llap_write_pending(llap->llap_page->mapping->host, llap);
GOTO(out, 0);
}
SIGNAL_MASK_ASSERT(); /* XXX BUG 1511 */
LASSERT(inode == file->f_dentry->d_inode);
LASSERT(PageLocked(page));
+ LASSERT(LLI_DIRTY_HANDLE(inode));
CDEBUG(D_INODE, "inode %p is writing page %p from %d to %d at %lu\n",
inode, page, from, to, page->index);
LASSERT(!PageDirty(page));
LASSERT(PageLocked(page));
+ LASSERT(LLI_DIRTY_HANDLE(inode));
exp = ll_i2dtexp(inode);
if (exp == NULL)
unlock_page(page);
- if (0 && cmd == OBD_BRW_WRITE) {
+ if (cmd == OBD_BRW_WRITE) {
llap_write_complete(page->mapping->host, llap);
ll_try_done_writing(page->mapping->host);
}
return;
}
- //llap_write_complete(inode, llap);
+ llap_write_complete(inode, llap);
rc = obd_teardown_async_page(exp, ll_i2info(inode)->lli_smd, NULL,
llap->llap_cookie);
if (rc != 0)
}
ll_och_fill(inode, it, och);
/* ll_md_och_close() will free och */
- ll_md_och_close(ll_i2mdexp(inode), inode, och);
+ ll_md_och_close(ll_i2mdexp(inode), inode, och, 0);
}
(*och_usecount)++;
{
struct obd_device *obd = exp->exp_obd;
struct lmv_obd *lmv = &obd->u.lmv;
- int rc = 0;
+ int i, rc;
ENTRY;
rc = lmv_check_connect(obd);
CDEBUG(D_OTHER, "CBDATA for "DLID4"\n", OLID4(id));
LASSERT(id_group(id) < lmv->desc.ld_tgt_count);
- rc = md_change_cbdata(lmv->tgts[id_group(id)].ltd_exp,
- id, it, data);
+ /* with CMD every object can have two locks in different
+ * namespaces: lookup lock in space of mds storing direntry
+ * and update/open lock in space of mds storing inode */
+ for (i = 0; i < lmv->desc.ld_tgt_count; i++)
+ md_change_cbdata(lmv->tgts[i].ltd_exp, id, it, data);
- RETURN(rc);
+ RETURN(0);
}
static int lmv_change_cbdata_name(struct obd_export *exp,
struct mds_body *body;
body = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*body));
- mdc_pack_id(&body->id1, oa->o_id, 0, oa->o_mode, 0, 0);
+ mdc_pack_id(&body->id1, oa->o_id, oa->o_generation, oa->o_mode, 0, 0);
memcpy(&body->handle, &och->och_fh, sizeof(body->handle));
if (oa->o_valid & OBD_MD_FLATIME) {
if (lock->l_ast_data && lock->l_ast_data != data) {
struct inode *new_inode = data;
struct inode *old_inode = lock->l_ast_data;
- LASSERTF(old_inode->i_state & I_FREEING,
- "Found existing inode %p/%lu/%u state %lu in lock: "
- "setting data to %p/%lu/%u\n", old_inode,
- old_inode->i_ino, old_inode->i_generation,
- old_inode->i_state, new_inode, new_inode->i_ino,
- new_inode->i_generation);
+ if (!(old_inode->i_state & I_FREEING)) {
+ CERROR("Found existing inode %p/%lu/%u state %lu "
+ "in lock: setting data to %p/%lu/%u\n",
+ old_inode, old_inode->i_ino,
+ old_inode->i_generation, old_inode->i_state,
+ new_inode, new_inode->i_ino,
+ new_inode->i_generation);
+ unlock_res_and_lock(lock);
+ LBUG();
+ }
}
#endif
lock->l_ast_data = data;
RETURN(0);
if (obd->obd_recovering) {
- CDEBUG(D_ERROR, "size for "DLID4" is unknown yet (recovering)\n",
+ CDEBUG(D_INODE, "size for "DLID4" is unknown yet (recovering)\n",
OLID4(&body->id1));
RETURN(0);
}
if (atomic_read(&inode->i_writecount)) {
/* some one has opened the file for write.
* mds doesn't know actual size */
- CDEBUG(D_OTHER, "MDS doesn't know actual size for "DLID4"\n",
+ CDEBUG(D_INODE, "MDS doesn't know actual size for "DLID4"\n",
OLID4(&body->id1));
RETURN(0);
}
- CDEBUG(D_OTHER, "MDS returns "LPD64"/"LPD64" for"DLID4"\n",
+ CDEBUG(D_INODE, "MDS returns "LPD64"/"LPD64" for"DLID4"\n",
body->size, body->blocks, OLID4(&body->id1));
- body->valid |= OBD_MD_FLSIZE;
+ body->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
RETURN(0);
}
(inode)->i_flags &= ~(0x4000000); \
CDEBUG(D_VFSTRACE, "removing orphan flag from inode %p\n", inode); \
} while (0)
+
+
+/* inode flags managed by mds directly */
+#define MDS_IF_ATTRS_OLD 0x8000000 /* inode needs attrs. refreshing */
+
+#define mds_inode_has_old_attrs(inode) ((inode)->i_flags & MDS_IF_ATTRS_OLD)
+#define mds_inode_set_attrs_old(inode) \
+do { \
+ (inode)->i_flags |= MDS_IF_ATTRS_OLD; \
+ CDEBUG(D_VFSTRACE, "setting attr.old flag on inode %p\n", inode);\
+} while (0)
+#define mds_inode_unset_attrs_old(inode) \
+do { \
+ (inode)->i_flags &= ~(MDS_IF_ATTRS_OLD); \
+ CDEBUG(D_VFSTRACE, "removing attrs.old flag from inode %p\n", inode);\
+} while (0)
+
+
#endif /* __KERNEL__ */
/* mds/mds_reint.c */
int unlink_orphan);
int mds_close(struct ptlrpc_request *req, int offset);
int mds_done_writing(struct ptlrpc_request *req, int offset);
-
+int mds_validate_size(struct obd_device *obd, struct inode *inode,
+ struct mds_body *body, struct iattr *iattr);
/* mds/mds_fs.c */
int mds_client_add(struct obd_device *obd, struct mds_obd *mds,
reply_body = lustre_msg_buf(req->rq_repmsg, 0,
sizeof(*reply_body));
+ if (request_body && (request_body->valid & OBD_MD_FLSIZE)) {
+ /* we set i_size/i_blocks here, nobody will see
+ * them until all write references are dropped.
+ * btw, we hold one reference */
+ LASSERT(mfd->mfd_mode & FMODE_WRITE);
+ i_size_write(inode, request_body->size);
+ inode->i_blocks = request_body->blocks;
+ iattr.ia_size = inode->i_size;
+ iattr.ia_valid |= ATTR_SIZE;
+ mds_inode_unset_attrs_old(inode);
+ }
+
idlen = ll_id2str(idname, inode->i_ino, inode->i_generation);
CDEBUG(D_INODE, "inode %p ino %s nlink %d orphan %d\n", inode,
idname, inode->i_nlink, mds_orphan_open_count(inode));
}
goto out; /* Don't bother updating attrs on unlinked inode */
- } else if ((mfd->mfd_mode & FMODE_WRITE) && rc == 0 && request_body) {
+ } else if ((mfd->mfd_mode & FMODE_WRITE) && rc == 0) {
/* last writer closed file - let's update i_size/i_blocks */
- if (request_body->valid & OBD_MD_FLSIZE) {
- LASSERT(request_body->valid & OBD_MD_FLBLOCKS);
- CDEBUG(D_OTHER, "update size "LPD64" for "DLID4
- ", epoch "LPD64"\n", inode->i_size,
- OLID4(&request_body->id1),
- request_body->io_epoch);
- iattr.ia_size = inode->i_size;
- iattr.ia_valid |= ATTR_SIZE;
- }
+ mds_validate_size(obd, inode, request_body, &iattr);
}
#if 0
/* If other clients have this file open for write, rc will be > 0 */
if (rc > 0)
rc = 0;
+ if (!obd->obd_recovering && mds_inode_has_old_attrs(inode)
+ && !mds_inode_is_orphan(inode)
+ && atomic_read(&inode->i_writecount) == 0) {
+ CERROR("leave inode %lu/%u with old attributes\n",
+ inode->i_ino, inode->i_generation);
+ }
l_dput(mfd->mfd_dentry);
mds_mfd_destroy(mfd);
__u64 lov_merge_size(struct lov_stripe_md *lsm, int kms);
__u64 lov_merge_blocks(struct lov_stripe_md *lsm);
-int mds_validate_size(struct obd_device *obd, struct mds_body *body,
- struct mds_file_data *mfd)
+int mds_validate_size(struct obd_device *obd, struct inode *inode,
+ struct mds_body *body, struct iattr *iattr)
{
ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };
- struct inode *inode = mfd->mfd_dentry->d_inode;
struct lustre_handle lockh = { 0 };
struct lov_stripe_md *lsm = NULL;
int rc, len, flags;
if (!S_ISREG(inode->i_mode))
RETURN(0);
- /* we update i_size/i_blocks only for writers */
- if (!(mfd->mfd_mode & FMODE_WRITE))
- RETURN(0);
-
- /* we like when client reports actual i_size/i_blocks himself */
- if (body->valid & OBD_MD_FLSIZE) {
- LASSERT(body->valid & OBD_MD_FLBLOCKS);
- CDEBUG(D_OTHER, "client reports "LPD64"/"LPD64" for "DLID4"\n",
- body->size, body->blocks, OLID4(&body->id1));
- RETURN(0);
- }
-
/* we shouldn't fetch size from OSTes during recovery - deadlock */
- if (obd->obd_recovering)
+ if (obd->obd_recovering) {
+ CERROR("size-on-mds has no support on OST yet\n");
RETURN(0);
+ }
- DOWN_READ_I_ALLOC_SEM(inode);
- if (atomic_read(&inode->i_writecount) > 1
- || mds_inode_is_orphan(inode)) {
- /* there is no need to update i_size/i_blocks on orphans.
- * also, if this is not last writer, then it doesn't make
- * sense to fetch i_size/i_blocks from OSSes */
- UP_READ_I_ALLOC_SEM(inode);
+ /* if nobody modified attrs. we're lucky */
+ if (!mds_inode_has_old_attrs(inode))
RETURN(0);
- }
- UP_READ_I_ALLOC_SEM(inode);
/* 1: client didn't send actual i_size/i_blocks
* 2: we seem to be last writer
GOTO(cleanup, rc);
}
- body->size = lov_merge_size(lsm, 0);
- body->blocks = lov_merge_blocks(lsm);
- body->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
+ CDEBUG(D_INODE, "LOV reports "LPD64"/%lu for "DLID4" [%s%s%s]\n",
+ inode->i_size, inode->i_blocks, OLID4(&body->id1),
+ atomic_read(&inode->i_writecount) > 1 ? "U" : "",
+ mds_inode_has_old_attrs(inode) ? "D" : "",
+ mds_inode_is_orphan(inode) ? "O" : "");
- CDEBUG(D_OTHER, "LOV reports "LPD64"/"LPD64" for "DLID4"\n",
- body->size, body->blocks, OLID4(&body->id1));
+ i_size_write(inode, lov_merge_size(lsm, 0));
+ inode->i_blocks = lov_merge_blocks(lsm);
+ iattr->ia_size = inode->i_size;
+ iattr->ia_valid |= ATTR_SIZE;
+ DOWN_WRITE_I_ALLOC_SEM(inode);
+ mds_inode_unset_attrs_old(inode);
+ UP_WRITE_I_ALLOC_SEM(inode);
obd_cancel(obd->u.mds.mds_dt_exp, lsm, LCK_PR, &lockh);
RETURN(-ESTALE);
}
- rc = mds_validate_size(obd, body, mfd);
- LASSERT(rc == 0);
-
inode = mfd->mfd_dentry->d_inode;
- if (mfd->mfd_mode & FMODE_WRITE) {
- /* we set i_size/i_blocks here, nobody will see
- * them until all write references are dropped.
- * btw, we hold one reference */
- if (body->valid & OBD_MD_FLSIZE)
- i_size_write(inode, body->size);
- if (body->valid & OBD_MD_FLBLOCKS)
- inode->i_blocks = body->blocks;
- }
-
/* child i_alloc_sem protects orphan_dec_test && is_orphan race */
DOWN_WRITE_I_ALLOC_SEM(inode); /* mds_mfd_close drops this */
+
+ if (body->flags & MDS_BFLAG_DIRTY_EPOCH) {
+ /* the client modified data through the handle
+ * we need to care about attrs. -bzzz */
+ mds_inode_set_attrs_old(inode);
+ }
+
if (mds_inode_is_orphan(inode) && mds_orphan_open_count(inode) == 1) {
struct mds_body *rep_body;
ost_stime_record(req, &start, 1, 2);
if (rc == 0) {
- repbody = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*repbody));
- memcpy(&repbody->oa, &body->oa, sizeof(repbody->oa));
-
#if CHECKSUM_BULK
repbody->oa.o_cksum = ost_checksum_bulk(desc);
repbody->oa.o_valid |= OBD_MD_FLCKSUM;
(svc->srv_nthreads - 1))),
&lwi);
+#if 0
+ /* disable watchdog: with CMD server can issue request
+ * to another server to satisfy the request -bzzz */
lc_watchdog_touch(watchdog);
+#endif
ptlrpc_check_rqbd_pools(svc);
if (!list_empty (&svc->srv_reply_queue))
struct fsfilt_operations *cache_fsfilt = I2FOPS(dentry->d_inode);
struct dentry *cache_dentry = NULL;
struct inode *cache_inode = I2CI(dentry->d_inode);
+ struct smfs_super_info *sbi = S2SMI(dentry->d_inode->i_sb);
struct hook_setattr_msg msg = {
.dentry = dentry,
.attr = iattr
SMFS_PRE_HOOK(dentry->d_inode, HOOK_F_SETATTR, &msg);
+ if (SMFS_DO_HND_IBLOCKS(sbi)) {
+ /* size-on-mds changes i_blocks directly to reflect
+ * aggregated i_blocks from all OSTs -bzzz */
+ cache_inode->i_blocks = dentry->d_inode->i_blocks;
+ }
rc = cache_fsfilt->fs_setattr(cache_dentry, handle, iattr, do_trunc);
SMFS_POST_HOOK(dentry->d_inode, HOOK_F_SETATTR, &msg, rc);
struct mds_obd * mds = &obd->u.mds;
smfs_mds_flags(mds, root_dentry->d_inode);
+ SMFS_SET_HND_IBLOCKS(smb);
}
else
CDEBUG(D_SUPER,"Unknown OBD (%s) post_setup\n",
# - 65h (default stripe inheritance) is not implemented for LMV
# configurations. Will be done in second phase of collibri.
# - 71 mmap still not updated on HEAD
+# - 42b (current implementation of size-on-mds feature doesn't handle this)
-ALWAYS_EXCEPT=${ALWAYS_EXCEPT:-"24n 48a 51b 51c 65h 71"}
+ALWAYS_EXCEPT=${ALWAYS_EXCEPT:-"24n 48a 51b 51c 65h 71 42b"}
# UPDATE THE COMMENT ABOVE WITH BUG NUMBERS WHEN CHANGING ALWAYS_EXCEPT!
[ "$ALWAYS_EXCEPT$EXCEPT" ] && echo "Skipping tests: $ALWAYS_EXCEPT $EXCEPT"
test_43a() {
mkdir -p $DIR/d43
cp -p `which multiop` $DIR/d43/multiop
+ sync
$DIR/d43/multiop $TMP/test43.junk O_c &
MULTIPID=$!
sleep 1
test_43b() {
mkdir -p $DIR/d43
cp -p `which multiop` $DIR/d43/multiop
+ sync
$DIR/d43/multiop $TMP/test43.junk O_c &
MULTIPID=$!
sleep 1
test_14a() {
mkdir -p $DIR1/d14
cp -p `which multiop` $DIR1/d14/multiop || error "cp failed"
+ sync
$DIR1/d14/multiop $TMP/test14.junk O_c &
MULTIPID=$!
sleep 1
git://git.whamcloud.com / fs / lustre-release.git / commitdiff