{

Client *client = static_cast<Client*>(p);

client->flush_set_callback(oset);

: inode(in), offset(0), this_offset(2), next_offset(2),

release_count(0), ordered_count(0), start_shared_gen(0),

buffer(0) {

inode->get();

: Dispatcher(m->cct),

logger(NULL),

m_command_hook(this),

timer(m->cct, client_lock),

callback_handle(NULL),

switch_interrupt_cb(NULL),

remount_cb(NULL),

ino_invalidate_cb(NULL),

dentry_invalidate_cb(NULL),

getgroups_cb(NULL),

async_ino_invalidator(m->cct),

async_dentry_invalidator(m->cct),

interrupt_finisher(m->cct),

remount_finisher(m->cct),

objecter_finisher(m->cct),

tick_event(NULL),

monclient(mc), messenger(m), whoami(m->get_myname().num()),

cap_epoch_barrier(0),

initialized(false), authenticated(false),

mounted(false), unmounting(false),

local_osd(-1), local_osd_epoch(0),

unsafe_sync_write(0),

client_lock("Client::client_lock")

{

monclient->set_messenger(m);

last_tid = 0;

last_flush_seq = 0;

cwd = NULL;

//

root = 0;

num_flushing_caps = 0;

_dir_vxattrs_name_size = _vxattrs_calcu_name_size(_dir_vxattrs);

_file_vxattrs_name_size = _vxattrs_calcu_name_size(_file_vxattrs);

lru.lru_set_max(cct->_conf->client_cache_size);

lru.lru_set_midpoint(cct->_conf->client_cache_mid);

// file handles

free_fd_set.insert(10, 1<<30);

// set up messengers

messenger = m;

// osd interfaces

mdsmap = new MDSMap;

objecter = new Objecter(cct, messenger, monclient, NULL,

0, 0);

objecter->set_client_incarnation(0); // client always 0, for now.

writeback_handler = new ObjecterWriteback(objecter, &objecter_finisher,

&client_lock);

objectcacher = new ObjectCacher(cct, "libcephfs", *writeback_handler, client_lock,

client_flush_set_callback, // all commit callback

(void*)this,

cct->_conf->client_oc_size,

cct->_conf->client_oc_max_objects,

cct->_conf->client_oc_max_dirty,

cct->_conf->client_oc_target_dirty,

cct->_conf->client_oc_max_dirty_age,

true);

objecter_finisher.start();

filer = new Filer(objecter, &objecter_finisher);

{

// fd's

for (ceph::unordered_map<int, Fh*>::iterator it = fd_map.begin();

it != fd_map.end();

++it) {

Fh *fh = it->second;

ldout(cct, 1) << "tear_down_cache forcing close of fh " << it->first << " ino " << fh->inode->ino << dendl;

put_inode(fh->inode);

delete fh;

}

fd_map.clear();

// caps!

// *** FIXME ***

// empty lru

lru.lru_set_max(0);

trim_cache();

assert(lru.lru_get_size() == 0);

// close root ino

assert(inode_map.size() <= 1 + root_parents.size());

if (root && inode_map.size() == 1 + root_parents.size()) {

delete root;

root = 0;

root_ancestor = 0;

while (!root_parents.empty()) {

Inode *in = root_parents.begin()->second;

root_parents.erase(root_parents.begin());

delete in;

}

inode_map.clear();

}

assert(inode_map.empty());

{

filepath path;

in->make_long_path(path);

ldout(cct, 1) << "dump_inode: "

<< (disconnected ? "DISCONNECTED ":"")

<< "inode " << in->ino

<< " " << path

<< " ref " << in->get_num_ref()

<< *in << dendl;

if (f) {

f->open_object_section("inode");

f->dump_stream("path") << path;

if (disconnected)

f->dump_int("disconnected", 1);

in->dump(f);

f->close_section();

}

did.insert(in);

if (in->dir) {

ldout(cct, 1) << " dir " << in->dir << " size " << in->dir->dentries.size() << dendl;

for (ceph::unordered_map<string, Dentry*>::iterator it = in->dir->dentries.begin();

it != in->dir->dentries.end();

++it) {

ldout(cct, 1) << " " << in->ino << " dn " << it->first << " " << it->second << " ref " << it->second->ref << dendl;

if (f) {

f->open_object_section("dentry");

it->second->dump(f);

f->close_section();

}

if (it->second->inode)

dump_inode(f, it->second->inode, did, false);

}

}

{

set<Inode*> did;

ldout(cct, 1) << "dump_cache" << dendl;

if (f)

f->open_array_section("cache");

if (root)

dump_inode(f, root, did, true);

// make a second pass to catch anything disconnected

for (ceph::unordered_map<vinodeno_t, Inode*>::iterator it = inode_map.begin();

it != inode_map.end();

++it) {

if (did.count(it->second))

continue;

dump_inode(f, it->second, did, true);

}

if (f)

f->close_section();

{

assert(client_lock.is_locked_by_me());

ldout(cct, 1) << __func__ << dendl;

const OSDMap *osdmap = objecter->get_osdmap_read();

const epoch_t osd_epoch = osdmap->get_epoch();

objecter->put_osdmap_read();

if (f) {

f->open_object_section("metadata");

{

for (std::map<std::string, std::string>::const_iterator i = metadata.begin();

i != metadata.end(); ++i) {

f->dump_string(i->first.c_str(), i->second);

}

}

f->close_section();

f->dump_int("dentry_count", lru.lru_get_size());

f->dump_int("dentry_pinned_count", lru.lru_get_num_pinned());

f->dump_int("inode_count", inode_map.size());

f->dump_int("mds_epoch", mdsmap->get_epoch());

f->dump_int("osd_epoch", osd_epoch);

f->dump_int("osd_epoch_barrier", cap_epoch_barrier);

}

{

client_lock.Lock();

assert(!initialized);

timer.init();

objectcacher->start();

objecter->init();

// ok!

messenger->add_dispatcher_tail(objecter);

messenger->add_dispatcher_tail(this);

int r = monclient->init();

if (r < 0) {

// need to do cleanup because we're in an intermediate init state

objecter->shutdown();

timer.shutdown();

client_lock.Unlock();

objectcacher->stop();

monclient->shutdown();

return r;

}

objecter->start();

monclient->set_want_keys(CEPH_ENTITY_TYPE_MDS | CEPH_ENTITY_TYPE_OSD);

monclient->sub_want("mdsmap", 0, 0);

monclient->renew_subs();

// logger

PerfCountersBuilder plb(cct, "client", l_c_first, l_c_last);

plb.add_time_avg(l_c_reply, "reply");

plb.add_time_avg(l_c_lat, "lat");

plb.add_time_avg(l_c_wrlat, "wrlat");

plb.add_time_avg(l_c_owrlat, "owrlat");

plb.add_time_avg(l_c_ordlat, "ordlat");

logger = plb.create_perf_counters();

cct->get_perfcounters_collection()->add(logger);

client_lock.Unlock();

AdminSocket* admin_socket = cct->get_admin_socket();

int ret = admin_socket->register_command("mds_requests",

"mds_requests",

&m_command_hook,

"show in-progress mds requests");

if (ret < 0) {

lderr(cct) << "error registering admin socket command: "

<< cpp_strerror(-ret) << dendl;

}

ret = admin_socket->register_command("mds_sessions",

"mds_sessions",

&m_command_hook,

"show mds session state");

if (ret < 0) {

lderr(cct) << "error registering admin socket command: "

<< cpp_strerror(-ret) << dendl;

}

ret = admin_socket->register_command("dump_cache",

"dump_cache",

&m_command_hook,

"show in-memory metadata cache contents");

if (ret < 0) {

lderr(cct) << "error registering admin socket command: "

<< cpp_strerror(-ret) << dendl;

}

ret = admin_socket->register_command("kick_stale_sessions",

"kick_stale_sessions",

&m_command_hook,

"kick sessions that were remote reset");

if (ret < 0) {

lderr(cct) << "error registering admin socket command: "

<< cpp_strerror(-ret) << dendl;

}

ret = admin_socket->register_command("status",

"status",

&m_command_hook,

"show overall client status");

if (ret < 0) {

lderr(cct) << "error registering admin socket command: "

<< cpp_strerror(-ret) << dendl;

}

populate_metadata();

client_lock.Lock();

initialized = true;

client_lock.Unlock();

return r;

{

ldout(cct, 1) << "shutdown" << dendl;

AdminSocket* admin_socket = cct->get_admin_socket();

admin_socket->unregister_command("mds_requests");

admin_socket->unregister_command("mds_sessions");

admin_socket->unregister_command("dump_cache");

admin_socket->unregister_command("kick_stale_sessions");

admin_socket->unregister_command("status");

if (ino_invalidate_cb) {

ldout(cct, 10) << "shutdown stopping cache invalidator finisher" << dendl;

async_ino_invalidator.wait_for_empty();

async_ino_invalidator.stop();

}

if (dentry_invalidate_cb) {

ldout(cct, 10) << "shutdown stopping dentry invalidator finisher" << dendl;

async_dentry_invalidator.wait_for_empty();

async_dentry_invalidator.stop();

}

if (switch_interrupt_cb) {

ldout(cct, 10) << "shutdown stopping interrupt finisher" << dendl;

interrupt_finisher.wait_for_empty();

interrupt_finisher.stop();

}

if (remount_cb) {

ldout(cct, 10) << "shutdown stopping remount finisher" << dendl;

remount_finisher.wait_for_empty();

remount_finisher.stop();

}

objectcacher->stop(); // outside of client_lock! this does a join.

client_lock.Lock();

assert(initialized);

initialized = false;

timer.shutdown();

objecter->shutdown();

client_lock.Unlock();

objecter_finisher.wait_for_empty();

objecter_finisher.stop();

monclient->shutdown();

if (logger) {

cct->get_perfcounters_collection()->remove(logger);

delete logger;

logger = NULL;

}

{

ldout(cct, 20) << "trim_cache size " << lru.lru_get_size() << " max " << lru.lru_get_max() << dendl;

unsigned last = 0;

while (lru.lru_get_size() != last) {

last = lru.lru_get_size();

if (lru.lru_get_size() <= lru.lru_get_max()) break;

// trim!

Dentry *dn = static_cast<Dentry*>(lru.lru_expire());

if (!dn)

break; // done

trim_dentry(dn);

}

// hose root?

if (lru.lru_get_size() == 0 && root && root->get_num_ref() == 0 && inode_map.size() == 1 + root_parents.size()) {

ldout(cct, 15) << "trim_cache trimmed root " << root << dendl;

delete root;

root = 0;

root_ancestor = 0;

while (!root_parents.empty()) {

Inode *in = root_parents.begin()->second;

root_parents.erase(root_parents.begin());

delete in;

}

inode_map.clear();

}

{

mds_rank_t mds = s->mds_num;

ldout(cct, 20) << "trim_cache_for_reconnect mds." << mds << dendl;

int trimmed = 0;

list<Dentry*> skipped;

while (lru.lru_get_size() > 0) {

Dentry *dn = static_cast<Dentry*>(lru.lru_expire());

if (!dn)

break;

if ((dn->inode && dn->inode->caps.count(mds)) ||

dn->dir->parent_inode->caps.count(mds)) {

trim_dentry(dn);

trimmed++;

} else

skipped.push_back(dn);

}

for(list<Dentry*>::iterator p = skipped.begin(); p != skipped.end(); ++p)

lru.lru_insert_mid(*p);

ldout(cct, 20) << "trim_cache_for_reconnect mds." << mds

<< " trimmed " << trimmed << " dentries" << dendl;

if (s->caps.size() > 0)

_invalidate_kernel_dcache();

{

ldout(cct, 15) << "trim_dentry unlinking dn " << dn->name

<< " in dir " << hex << dn->dir->parent_inode->ino

<< dendl;

dn->dir->release_count++;

if (dn->dir->parent_inode->flags & I_COMPLETE) {

ldout(cct, 10) << " clearing (I_COMPLETE|I_DIR_ORDERED) on " << *dn->dir->parent_inode << dendl;

dn->dir->parent_inode->flags &= ~(I_COMPLETE | I_DIR_ORDERED);

}

unlink(dn, false, false); // drop dir, drop dentry

uint64_t truncate_seq, uint64_t truncate_size,

uint64_t size,

uint64_t time_warp_seq, utime_t ctime,

utime_t mtime,

utime_t atime,

version_t inline_version,

bufferlist& inline_data,

int issued)

{

bool warn = false;

ldout(cct, 10) << "update_inode_file_bits " << *in << " " << ccap_string(issued)

<< " mtime " << mtime << dendl;

ldout(cct, 25) << "truncate_seq: mds " << truncate_seq << " local "

<< in->truncate_seq << " time_warp_seq: mds " << time_warp_seq

<< " local " << in->time_warp_seq << dendl;

uint64_t prior_size = in->size;

if (inline_version > in->inline_version) {

in->inline_data = inline_data;

in->inline_version = inline_version;

}

if (truncate_seq > in->truncate_seq ||

(truncate_seq == in->truncate_seq && size > in->size)) {

ldout(cct, 10) << "size " << in->size << " -> " << size << dendl;

in->size = size;

in->reported_size = size;

if (truncate_seq != in->truncate_seq) {

ldout(cct, 10) << "truncate_seq " << in->truncate_seq << " -> "

<< truncate_seq << dendl;

in->truncate_seq = truncate_seq;

in->oset.truncate_seq = truncate_seq;

// truncate cached file data

if (prior_size > size) {

_invalidate_inode_cache(in, truncate_size, prior_size - truncate_size);

}

}

// truncate inline data

if (in->inline_version < CEPH_INLINE_NONE) {

uint32_t len = in->inline_data.length();

if (size < len)

in->inline_data.splice(size, len - size);

}

}

if (truncate_seq >= in->truncate_seq &&

in->truncate_size != truncate_size) {

if (in->is_file()) {

ldout(cct, 10) << "truncate_size " << in->truncate_size << " -> "

<< truncate_size << dendl;

in->truncate_size = truncate_size;

in->oset.truncate_size = truncate_size;

} else {

ldout(cct, 0) << "Hmmm, truncate_seq && truncate_size changed on non-file inode!" << dendl;

}

}

// be careful with size, mtime, atime

if (issued & (CEPH_CAP_FILE_EXCL|

CEPH_CAP_FILE_WR|

CEPH_CAP_FILE_BUFFER|

CEPH_CAP_AUTH_EXCL|

CEPH_CAP_XATTR_EXCL)) {

ldout(cct, 30) << "Yay have enough caps to look at our times" << dendl;

if (ctime > in->ctime)

in->ctime = ctime;

if (time_warp_seq > in->time_warp_seq) {

ldout(cct, 10) << "mds time_warp_seq " << time_warp_seq << " on inode " << *in

<< " is higher than local time_warp_seq "

<< in->time_warp_seq << dendl;

//the mds updated times, so take those!

in->mtime = mtime;

in->atime = atime;

in->time_warp_seq = time_warp_seq;

} else if (time_warp_seq == in->time_warp_seq) {

//take max times

if (mtime > in->mtime)

in->mtime = mtime;

if (atime > in->atime)

in->atime = atime;

} else if (issued & CEPH_CAP_FILE_EXCL) {

//ignore mds values as we have a higher seq

} else warn = true;

} else {

ldout(cct, 30) << "Don't have enough caps, just taking mds' time values" << dendl;

if (time_warp_seq >= in->time_warp_seq) {

in->ctime = ctime;

in->mtime = mtime;

in->atime = atime;

in->time_warp_seq = time_warp_seq;

} else warn = true;

}

if (warn) {

ldout(cct, 0) << "WARNING: " << *in << " mds time_warp_seq "

<< time_warp_seq << " is lower than local time_warp_seq "

<< in->time_warp_seq

<< dendl;

}

{

for (map<frag_t,int>::iterator p = in->fragmap.begin(); p != in->fragmap.end(); )

if (!in->dirfragtree.is_leaf(p->first))

in->fragmap.erase(p++);

else

++p;

{

utime_t dttl = from;

dttl += (float)dlease->duration_ms / 1000.0;

assert(dn && dn->inode);

if (dlease->mask & CEPH_LOCK_DN) {

if (dttl > dn->lease_ttl) {

ldout(cct, 10) << "got dentry lease on " << dn->name

<< " dur " << dlease->duration_ms << "ms ttl " << dttl << dendl;

dn->lease_ttl = dttl;

dn->lease_mds = session->mds_num;

dn->lease_seq = dlease->seq;

dn->lease_gen = session->cap_gen;

}

}

dn->cap_shared_gen = dn->dir->parent_inode->shared_gen;

{

// auth

ldout(cct, 20) << "got dirfrag map for " << in->ino << " frag " << dst->frag << " to mds " << dst->auth << dendl;

if (dst->auth >= 0) {

in->fragmap[dst->frag] = dst->auth;

} else {

in->fragmap.erase(dst->frag);

}

if (!in->dirfragtree.is_leaf(dst->frag)) {

in->dirfragtree.force_to_leaf(cct, dst->frag);

_fragmap_remove_non_leaves(in);

}

// replicated

in->dir_replicated = !dst->dist.empty(); // FIXME that's just one frag!

// dist

/*

MClientReply *reply = request->reply;

ConnectionRef con = request->reply->get_connection();

uint64_t features = con->get_features();

assert(request->readdir_result.empty());

// the extra buffer list is only set for readdir and lssnap replies

bufferlist::iterator p = reply->get_extra_bl().begin();

if (!p.end()) {

// snapdir?

if (request->head.op == CEPH_MDS_OP_LSSNAP) {

assert(diri);

diri = open_snapdir(diri);

}

// only open dir if we're actually adding stuff to it!

Dir *dir = diri->open_dir();

assert(dir);

// dirstat