gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- // This is for FUSE support of passing the userid
- //Ticket *tk;
- //if (uid == -1 || gid == -1)
- // tk = get_user_ticket(getuid(), getgid());
- //else
- // tk = get_user_ticket(uid, gid);
+ //Ticket *tk = get_user_ticket(uid, gid);
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
// main path arg is new link name
delete reply;
dout(10) << "link result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
+ //Ticket *tk = get_user_ticket(uid, gid);
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
dout(10) << "unlink result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
+ //Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string absfrom;
mkabspath(relfrom, absfrom);
delete reply;
dout(10) << "rename result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
+ //Ticket *tk = get_user_ticket(uid, gid);
//Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
dout(10) << "mkdir result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
+ //Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
dout(10) << "rmdir result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abstarget;
mkabspath(reltarget, abstarget);
delete reply;
dout(10) << "symlink result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
if (res > size) res = size;
memcpy(buf, in->symlink->c_str(), res);
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res; // return length in bytes (to mimic the system call)
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
dout(10) << "stat sez size = " << in->inode.size << " ino = " << stbuf->st_ino << endl;
}
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
dout(10) << "stat sez size = " << in->inode.size << " ino = " << in->inode.ino << endl;
}
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
dout(10) << "chmod result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
dout(10) << "chown result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
dout(10) << "utime result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
return res;
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
delete reply; //fix thing above first
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return res;
}
uid = getuid();
gid = getgid();
}
+
+ //Ticket *tk = get_user_ticket(uid, gid);
- Ticket *tk = get_user_ticket(uid, gid);
-
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
string abspath;
mkabspath(relpath, abspath);
const char *path = abspath.c_str();
+
+ // client access prediction stuff
+ /*
+ inodeno_t this_inode = path_map[abspath];
+ cout << "This inode is " << this_inode << endl;
+
+ if (this_inode != inodeno_t() && successor_inode[uid] != inodeno_t())
+ predicter[uid].add_observation(this_inode, successor_inode[uid]);
+
+ inodeno_t prediction = predicter[uid].predict_successor(this_inode);
- // note the successor relationship
- predicter[uid].add_observation(abspath, successor[uid]);
- string prediction = predicter[uid].predict_successor(abspath);
- successor[uid] = abspath;
- if (prediction.size() == 0)
+ if (prediction == inodeno_t()) {
cout << "Could not make confident prediction" << endl;
- else
+ }
+ else {
cout << "Predicted access of " << prediction << endl;
+ }
+ */
dout(3) << "op: fh = client->open(\"" << path << "\", " << flags << ");" << endl;
tout << "open" << endl;
// FIXME where does FUSE maintain user information
req->set_caller_uid(uid);
req->set_caller_gid(gid);
-
+
+ // add prediction info
+ //string pred_string;
+ //if (prediction != inodeno_t()) {
+ //char convertInode[sizeof(prediction)];
+ //memcpy(convertInode, &prediction, sizeof(convertInode));
+ //pred_string = string(convertInode, 8);
+ //}
+
MClientReply *reply = make_request(req, tryauth); // try auth if writer
-
+
assert(reply);
dout(3) << "op: open_files[" << reply->get_result() << "] = fh; // fh = " << reply->get_result() << endl;
tout << reply->get_result() << endl;
assert(f->inode);
f->inode->get();
+ // note successor relationships
+ //if (this_inode == inodeno_t() && successor_inode[uid] != inodeno_t())
+ //predicter[uid].add_observation(reply->get_ino(), successor_inode[uid]);
+
+ // update successor
+ //path_map[abspath] = reply->get_ino();
+ //successor_inode[uid] = reply->get_ino();
+
if (cmode & FILE_MODE_R) f->inode->num_open_rd++;
if (cmode & FILE_MODE_W) f->inode->num_open_wr++;
if (cmode & FILE_MODE_LAZY) f->inode->num_open_lazy++;
delete reply;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
trim_cache();
client_lock.Unlock();
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
dout(3) << "op: client->close(open_files[ " << fh << " ]);" << endl;
dout(3) << "op: open_files.erase( " << fh << " );" << endl;
put_inode( in );
int result = 0;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return result;
}
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
dout(3) << "op: client->read(" << fh << ", buf, " << size << ", " << offset << "); // that's " << offset << "~" << size << endl;
tout << "read" << endl;
dout(10) << "read rvalue " << rvalue << ", r " << r << endl;
// done!
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return rvalue;
}
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
//dout(7) << "write fh " << fh << " size " << size << " offset " << offset << endl;
dout(3) << "op: client->write(" << fh << ", buf, " << size << ", " << offset << ");" << endl;
in->file_wr_mtime = in->inode.mtime = g_clock.gettime();
// ok!
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return totalwritten;
}
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
dout(3) << "op: client->truncate(\"" << file << "\", " << size << ");" << endl;
tout << "truncate" << endl;
dout(10) << " truncate result is " << res << endl;
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return res;
}
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
dout(3) << "op: client->fsync(open_files[ " << fh << " ], " << syncdataonly << ");" << endl;
tout << "fsync" << endl;
while (!done) cond.Wait(client_lock);
}
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return r;
}
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
dout(3) << "op: client->lazyio_propogate(" << fd
<< ", " << offset << ", " << count << ")" << endl;
}
}
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return 0;
}
gid = getgid();
}
- Ticket *tk = get_user_ticket(uid, gid);
- //Ticket *tk = get_user_ticket(getuid(), getgid());
- if (!tk) {
- client_lock.Unlock();
- return -EPERM;
- }
+ //Ticket *tk = get_user_ticket(uid, gid);
+
+ //if (!tk) {
+ // client_lock.Unlock();
+ // return -EPERM;
+ //}
dout(3) << "op: client->lazyio_synchronize(" << fd
<< ", " << offset << ", " << count << ")" << endl;
}
}
- put_user_ticket(tk);
+ //put_user_ticket(tk);
client_lock.Unlock();
return 0;
}
// prediction
map<uid_t, string > successor;
+ map<uid_t, inodeno_t> successor_inode;
+ map<string, inodeno_t> path_map;
map<uid_t, RecentPopularity> predicter;
Ticket *get_user_ticket(uid_t uid, gid_t gid);
fix_client_id: 0, /* 0=off, 1=on */
renewal: 0, /* 0=off, 1=on */
renewal_period: 240, /* renew every 4 minutes */
+ config_predict: 0, /* 0=off, non-zero = filename ptr */
+ collect_predictions: 0, /* 0=off, 1=on */
client_aux: 0, /* 0=off, 1=on */
sign_scheme: 0, /* 0=esign, 1=RSA */
hash_scheme: 0, /* 0=sha-1, 1=sha-256,
g_conf.renewal = atoi(args[++i]);
else if (strcmp(args[i], "--renewal_period") == 0)
g_conf.renewal_period = atoi(args[++i]);
+ else if (strcmp(args[i], "--config_predict") == 0)
+ g_conf.config_predict = args[++i];
+ else if (strcmp(args[i], "--collect_predictions") == 0)
+ g_conf.collect_predictions = atoi(args[++i]);
else if (strcmp(args[i], "--file_layout_ssize") == 0)
g_OSD_FileLayout.stripe_size = atoi(args[++i]);
int fix_client_id;
int renewal;
int renewal_period;
+ char* config_predict;
+ int collect_predictions;
int client_aux;
int sign_scheme;
int hash_scheme;
//FIXME need to re-compute hash
}
+ int num_inodes() { return inodes.size(); }
+
bool contains(uid_t user) {
for (list<uid_t>::iterator ui = users.begin();
ui != users.end();
int J;
// sequence size parameter (6 is a happy default)
int K;
- map < string, deque<string> > inode_sequences;
+ map < inodeno_t, deque<inodeno_t> > inode_sequences;
public:
RecentPopularity() : J(4), K(6) {}
RecentPopularity(int jay, int kay) : J(jay), K(kay) {}
+ RecentPopularity(map<inodeno_t, deque<inodeno_t> > sequence) : J(4), K(6),
+ inode_sequences(sequence) {}
- void add_observation(string X, string successor) {
+
+ map<inodeno_t, deque<inodeno_t> >& get_sequence() { return inode_sequences; }
+ void add_observation(inodeno_t X, inodeno_t successor) {
inode_sequences[X].push_back(successor);
if (inode_sequences[X].size() > (unsigned)K)
inode_sequences[X].pop_front();
}
- string predict_successor(string X) {
+ inodeno_t predict_successor(inodeno_t X) {
+
+ //debug -- remove this at some point
+ //for (deque<inodeno_t>::reverse_iterator test_it = inode_sequences[X].rbegin();
+ //test_it != inode_sequences[X].rend();
+ //test_it++) {
+ //cout << *test_it << endl;
+ //}
// is our known sequence big enough?
if (inode_sequences[X].size() < (unsigned)K)
- return string();
+ return inodeno_t();
// can we make a prediction with confidence?
- set<string> checked_inodes;
+ set<inodeno_t> checked_inodes;
unsigned int index = 0;
- for (deque<string>::reverse_iterator iri = inode_sequences[X].rbegin();
+ for (deque<inodeno_t>::reverse_iterator iri = inode_sequences[X].rbegin();
iri != inode_sequences[X].rend();
iri++) {
// are there enough unchecked inodes to even keep going?
if (inode_sequences[X].size() - index >= (unsigned)J) {
int occurance = 0;
- for (deque<string>::reverse_iterator ini = iri;
+ for (deque<inodeno_t>::reverse_iterator ini = iri;
ini != inode_sequences[X].rend();
ini++) {
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#ifndef __USERBATCH_H
+#define __USERBATCH_H
+
+#include "include/types.h"
+#include "common/Clock.h"
+#include "common/Mutex.h"
+#include "common/Cond.h"
+
+#include "messages/MClientRequest.h"
+//#include "mds/Server.h"
+
+class Server;
+//class MDS;
+
+class UserBatch {
+
+ Server *server;
+ MDS *mds;
+
+ // is the thread initiated
+ bool thread_init;
+ // is the thread running
+ bool batch_stop;
+ // are we batching requests
+ bool batching;
+
+ Mutex batch_lock;
+ Cond batch_cond;
+
+public:
+ utime_t one_req_ago;
+ utime_t two_req_ago;
+ uid_t user;
+ gid_t user_group;
+
+ set<MClientRequest*> batched_requests;
+ bool batch_id_set;
+ cap_id_t batch_id;
+
+ class BatchThread : public Thread {
+ UserBatch *batch;
+ public:
+ BatchThread() {}
+ BatchThread (UserBatch *ub) : batch (ub) {}
+ void *entry() {
+ batch->batch_entry();
+ return 0;
+ }
+ } batch_thread;
+
+ UserBatch(Server *serve, MDS* metads) : server(serve), mds(metads) {
+ thread_init = false;
+ batching = false;
+ batch_stop = false;
+ batch_thread = BatchThread(this);
+ batch_thread.create();
+ }
+ ~UserBatch () {
+ batch_lock.Lock();
+ batch_stop = true;
+ batch_cond.Signal();
+ batch_lock.Unlock();
+ batch_thread.join();
+ }
+
+ bool is_batching() { return batching; }
+ bool should_batch(utime_t new_request_time) {
+ //if (new_request_time - two_req_ago < utime_t(0,10000)) // 10ms between
+ if (new_request_time > utime_t()) // always batch
+ return true;
+ return false;
+ }
+ void update_batch_time(utime_t new_request_time) {
+ two_req_ago = one_req_ago;
+ one_req_ago = new_request_time;
+ }
+
+ void add_to_batch(MClientRequest *req) {
+ dout(1) << "Batching the request for uid:"
+ << req->get_caller_uid() << " on client:"
+ << req->get_client() << " for file:"
+ << req->get_ino() << " with client inst:"
+ << req->get_client_inst() << endl;
+
+ batch_lock.Lock();
+
+ // wait until the thread has initialized
+ while (! thread_init)
+ batch_cond.Wait(batch_lock);
+
+ // was batching thread already on?
+ if (batching) {
+ batched_requests.insert(req);
+ }
+ else {
+ // set the user were batching for
+ user = req->get_caller_uid();
+ user_group = req->get_caller_gid();
+
+ // set the future capid
+ batch_id.cid = mds->cap_id_count;
+ batch_id.mds_id = mds->get_nodeid();
+ mds->cap_id_count++;
+
+ batching = true;
+ batch_id_set = true;
+
+ batched_requests.insert(req);
+
+ // start the buffering now
+ batch_cond.Signal();
+ }
+
+ batch_lock.Unlock();
+ return;
+ }
+
+ void batch_entry()
+ {
+ cout << "batch thread start------>" << endl;
+ batch_lock.Lock();
+
+ // init myself and signal anyone waiting for me to init
+ thread_init = true;
+ batch_cond.Signal();
+
+ while(!batch_stop) {
+
+ // ifwe're not buffering, then,
+ // were gonna get signaled when we start buffering
+ // plus i need to release the lock for anyone
+ // waiting for me to init
+ while (!batching)
+ batch_cond.Wait(batch_lock);
+
+ // the sleep releases the lock and allows the dispatch
+ // to insert requests into the buffer
+ // sleep first, then serve cap
+ batch_cond.WaitInterval(catch_lock, utime_t(5,0));
+
+ // now i've slept, make cap for users
+ list<inodeno_t> inode_list;
+ CapGroup inode_hash;
+ for (set<MClientRequest *>::iterator si = buffered_reqs.begin();
+ si != buffered_reqs.end();
+ si++) {
+ inode_list.push_back((*ii)->get_ino());
+ inode_hash.add_inode((*ii)->get_ino());
+ }
+ inode_hash.sign_list(mds->getPrvKey());
+ mds->unix_groups_byhash[inode_hash.get_root_hash()]= inode_hash;
+
+ ExtCap *ext_cap = new ExtCap(FILE_MODE_RW,
+ user,
+ user_group,
+ inode_hash.get_root_hash());
+ ext_cap->set_type(USER_BATCH);
+ ext_cap->set_id(batch_id);
+ ext_cap->sign_extcap(mds->getPrvKey());
+
+ // put the cap every inodes cache
+ CInode *inode_cache;
+ for (list<inodeno_t>::iterator ili = inode_list.begin();
+ ili != inode_list.end();
+ ili++) {
+ inode_cache= mds->mdcache->inode_map[(*ili)];
+ inode_cache->add_user_extcap(user, ext_cap);
+ }
+
+ // let requests loose
+ for (set<MClientRequest *>::iterator ri = batched_requests.begin();
+ ri != batched_requests.end();
+ ri++) {
+ server->handle_client_open(*ri, this);
+ }
+
+ batched_requests.clear();
+
+ //turn batching off
+ batching = false;
+ }
+
+ batch_lock.Unlock();
+ cout << "<------batcher thread finish" << endl;
+ }
+
+};
+
+#endif
* Author: Andrew Leung Nov., 2006
******************************/
#include"CryptoLib.h"
-#include "MerkleTree.h"
+//#include "MerkleTree.h"
#include<iostream>
using namespace std;
// message to hash
const byte* msg = (const byte*)"hash me";
+ /*
// test merkle trees
MerkleTree mtree;
uid_t user1 = 1000;
cout << "mtree2 is bigger" << endl;
else
cout << "mtree is bigger" << endl;
+ */
// sha-1
byte digest[SHA1DIGESTSIZE];
toHex(digestmd5, hexmd5, MD5DIGESTSIZE, 2*MD5DIGESTSIZE);
cout << "MD5 of " << msg << " is " << string((const char*)hexmd5,2*MD5DIGESTSIZE) << endl;
+
// esign signature
byte* signMsg = (byte *)"Message to sign is getting bigger by the minutefdsfdfdsffdfsfsdfdsfdfdsfsdfdsfsdfdsfsdsfdssdfsdfdsfdsffds";
//char* keyInput = "esig1536.dat";
esignPriv privKey = esignPrivKey(keyInput);
esignPub pubKey = esignPubKey(privKey);
SigBuf mySignature = esignSig(signMsg, strlen((const char*)signMsg), privKey);
+
// testing --> remove me!
byte testBuf[mySignature.size()];
memcpy((void*)testBuf,(void*)mySignature, mySignature.size());
else
cout << "Array copy Key failed" << endl;
+ /*
+
// RSA signature
byte* rsaMsg = (byte *)"Message to sign";
char* rsaInput = "rsa1024.dat";
decryptRC5(cipherRC5, plainRC5len, recoverRC5, decRC5);
cout << "My recovered message is " << recoverRC5 << endl;
+ */
+
return 0;
}
else
ext_cap = in->get_unix_world_cap();
}
+ if (g_conf.mds_group == 4) {
+ if (mds->predict_cap_cache[in->ino()].count(my_user) == 0)
+ ext_cap = 0;
+ else
+ ext_cap = &(mds->predict_cap_cache[in->ino()][my_user]);
+ }
// no grouping
else
ext_cap = in->get_user_extcap(my_user);
ext_cap->set_type(1);
+ }
+ // do prediction
+ else if (g_conf.mds_group == 4) {
+ // can we make any predictions?
+ if (mds->precompute_succ.count(in->ino()) != 0) {
+ cout << "Making a prediction in capability for " << in->ino() << endl;
+ // add the hash
+ hash_t inode_hash = mds->precompute_succ[in->ino()];
+ ext_cap = new ExtCap(FILE_MODE_RW, my_user, my_group, inode_hash);
+ ext_cap->set_type(USER_BATCH);
+ }
+ else {
+ cout << "Can't make predictions for this cap for " << in->ino() << endl;
+ ext_cap = new ExtCap(my_want, my_user, in->ino());
+ ext_cap->set_type(0);
+ }
+
}
// default no grouping
else {
else
in->set_unix_world_cap(ext_cap);
}
+ else if (g_conf.mds_group == 4) {
+ // did we use a hash for the inodes?
+ if (ext_cap->get_type() == USER_BATCH) {
+ hash_t inode_hash = ext_cap->get_file_hash();
+ list<inodeno_t> inode_list = mds->unix_groups_byhash[inode_hash].get_inode_list();
+ // cache in every inode included in the cap
+ for (list<inodeno_t>::iterator lii = inode_list.begin();
+ lii != inode_list.end();
+ lii++) {
+ mds->predict_cap_cache[*lii][my_user] = (*ext_cap);
+ }
+ }
+ mds->predict_cap_cache[in->ino()][my_user] = (*ext_cap);
+ }
else
in->add_user_extcap(my_user,ext_cap);
myPrivKey = esignPrivKey("crypto/esig1023.dat");
myPubKey = esignPubKey(myPrivKey);
+
// create unix_groups from file?
if (g_conf.unix_group_file) {
ifstream from(g_conf.unix_group_file);
}
}
+ // do prediction read-in
+ if (g_conf.mds_group == 4) {
+ map<inodeno_t, deque<inodeno_t> > sequence;
+ int off = 0;
+ bufferlist bl;
+ server->get_bl_ss(bl);
+ ::_decode(sequence, bl, off);
+ rp_predicter = RecentPopularity(sequence);
+
+ for (map<inodeno_t, deque<inodeno_t> >::iterator mi = sequence.begin();
+ mi != sequence.end();
+ mi++) {
+ CapGroup inode_list;
+ inodeno_t prediction;
+ prediction = rp_predicter.predict_successor(mi->first);
+
+ cout << "Predictions for " << mi->first << ": ";
+ while(prediction != inodeno_t()) {
+ cout << prediction << ", ";
+ inode_list.add_inode(prediction);
+ prediction = rp_predicter.predict_successor(prediction);
+
+ }
+ cout << "Cannot make any further predictions" << endl;
+
+ // cache the list
+ if (inode_list.num_inodes() != 0) {
+ inode_list.sign_list(myPrivKey);
+ unix_groups_byhash[inode_list.get_root_hash()] = inode_list;
+ precompute_succ[mi->first] = inode_list.get_root_hash();
+ }
+ }
+ }
+
// cap identifiers
cap_id_count = 0;
#include "crypto/CapGroup.h"
#include "crypto/MerkleTree.h"
#include "crypto/Renewal.h"
+#include "crypto/RecentPopularity.h"
#define MDS_PORT_MAIN 0
#define MDS_PORT_SERVER 1
// batched file groups by user
map<uid_t, UserBatch*> user_batch;
+ // successors
+ RecentPopularity rp_predicter;
+ inodeno_t last_access;
+ map<inodeno_t, hash_t> precompute_succ;
+ map<inodeno_t , map<uid_t, ExtCap> > predict_cap_cache;
+
void queue_waitfor_active(Context *c) { waitfor_active.push_back(c); }
bool is_dne() { return state == MDSMap::STATE_DNE; }
new EString("truncate fixme"));
}
+set<inodeno_t> Server::parse_predictions(string pred_string) {
+ // get number of predictions
+ int first_index = pred_string.find_first_of(",");
+ int num_preds = atoi(pred_string.substr(0, first_index).c_str());
+ cout << "Number of predictions to make " << num_preds << endl;
+
+ string str_to_parse = pred_string.substr(first_index+1, pred_string.size()-1);
+ cout << "Remaining string " << str_to_parse << ":" << endl;
+
+ set<inodeno_t> successors;
+ string pred;
+ inodeno_t place_holder_inode;
+ int comma_index;
+ for (int parser = 0; parser < num_preds; parser++) {
+
+ comma_index = str_to_parse.find_first_of(",");
+ pred = str_to_parse.substr(0, comma_index);
+ cout << "Got token " << pred << ":" << endl;
+ if (pred == ";")
+ break;
+
+ memset((byte*)&place_holder_inode, 0x00, sizeof(place_holder_inode));
+ memcpy((byte*)&place_holder_inode, pred.c_str(), pred.size());
+ successors.insert(place_holder_inode);
+ cout << "Just inserted place_holder_inode " << place_holder_inode << endl;
+ str_to_parse = str_to_parse.substr(comma_index+1, str_to_parse.size()-1);
+ }
+
+ return successors;
+}
+
+int Server::get_bl_ss(bufferlist& bl)
+{
+ string dir = "/home/aleung/ssrc/ceph/branches/aleung/security1/ceph";
+ /*
+ char fn[200];
+ if (b) {
+ sprintf(fn, "%s/%s/%s", dir.c_str(), a, b);
+ } else {
+ sprintf(fn, "%s/%s", dir.c_str(), a);
+ }
+ */
+ char fn[200];
+ sprintf(fn, "predictions");
+
+ int fd = ::open(fn, O_RDONLY);
+ if (!fd) {
+ /*
+ if (b) {
+ dout(15) << "get_bl " << a << "/" << b << " DNE" << endl;
+ } else {
+ dout(15) << "get_bl " << a << " DNE" << endl;
+ }
+ */
+ dout(15) << "get_bl predictions DNE" << endl;
+ return 0;
+ }
+ // get size
+ struct stat st;
+ int rc = ::fstat(fd, &st);
+ assert(rc == 0);
+ __int32_t len = st.st_size;
+
+ // read buffer
+ bl.clear();
+ bufferptr bp(len);
+ int off = 0;
+ while (off < len) {
+ dout(20) << "reading at off " << off << " of " << len << endl;
+ int r = ::read(fd, bp.c_str()+off, len-off);
+ if (r < 0) derr(0) << "errno on read " << strerror(errno) << endl;
+ assert(r>0);
+ off += r;
+ }
+ bl.append(bp);
+ ::close(fd);
+
+ /*
+ if (b) {
+ dout(15) << "get_bl " << a << "/" << b << " = " << bl.length() << " bytes" << endl;
+ } else {
+ dout(15) << "get_bl " << a << " = " << bl.length() << " bytes" << endl;
+ }
+ */
+ dout(15) << "get_bl predictions = " << bl.length() << " bytes" << endl;
+
+ return len;
+}
+
+int Server::put_bl_ss(bufferlist& bl)
+{
+ string dir = "/home/aleung/ssrc/ceph/branches/aleung/security1/ceph";
+ /*
+ char fn[200];
+ sprintf(fn, "%s/%s", dir.c_str(), a);
+ if (b) {
+ ::mkdir(fn, 0755);
+ dout(15) << "put_bl " << a << "/" << b << " = " << bl.length() << " bytes" << endl;
+ sprintf(fn, "%s/%s/%s", dir.c_str(), a, b);
+ } else {
+ dout(15) << "put_bl " << a << " = " << bl.length() << " bytes" << endl;
+ }
+
+ char tfn[200];
+ sprintf(tfn, "%s.new", fn);
+ */
+ char tfn[200];
+ sprintf(tfn, "predictions");
+ int fd = ::open(tfn, O_WRONLY|O_CREAT);
+ assert(fd);
+
+ // chmod
+ ::fchmod(fd, 0644);
+
+ // write data
+ for (list<bufferptr>::const_iterator it = bl.buffers().begin();
+ it != bl.buffers().end();
+ it++) {
+ int r = ::write(fd, it->c_str(), it->length());
+ if (r != (int)it->length())
+ derr(0) << "put_bl_ss ::write() returned " << r << " not " << it->length() << endl;
+ if (r < 0)
+ derr(0) << "put_bl_ss ::write() errored out, errno is " << strerror(errno) << endl;
+ }
+
+ ::fsync(fd);
+ ::close(fd);
+ //::rename(tfn, fn);
+
+ return 0;
+}
// ===========================
// open, openc, close
return;
}
+
+ // check file prediction stuff
+ if (g_conf.collect_predictions != 0) {
+ // if this isnt the first access
+ if (mds->last_access != inodeno_t() && mds->last_access != cur->ino()) {
+ cout << "Observed " << cur->ino() << " followed "
+ << mds->last_access << endl;
+
+ mds->rp_predicter.add_observation(mds->last_access, cur->ino());
+ bufferlist bl;
+ ::_encode(mds->rp_predicter.get_sequence(), bl);
+ put_bl_ss(bl);
+ }
+
+ mds->last_access = cur->ino();
+ }
+
// auth for write access
// redirect the write open to the auth?
if (mode != FILE_MODE_R && mode != FILE_MODE_LAZY &&
CInode **pin, CDentry **pdn,
bool okexist=false);
+ // prediction stuff
+ set<inodeno_t> parse_predictions(string pred_string);
+ int put_bl_ss(bufferlist& bl);
+ int get_bl_ss(bufferlist& bl);
+
};
uid_t uid;
gid_t gid;
string pubKey;
+ //char userPublicKey[ESIGNKEYSIZE];
public:
MClientAuthUser() : Message(MSG_CLIENT_AUTH_USER) {}
MClientAuthUser(string un, uid_t u, gid_t g, string k) :
Message(MSG_CLIENT_AUTH_USER), username(un), uid(u), gid(g), pubKey(k) { }
+ //MClientAuthUser(string un, uid_t u, gid_t g, byte* k) :
+ // Message(MSG_CLIENT_AUTH_USER), username(un), uid(u), gid(g) {
+ // memset(userPublicKey, k, ESIGNKEYSIZE);
+ //}
char *get_type_name() { return "client_auth_user"; }
const string& get_str_key() { return pubKey; }
string sarg;
string sarg2;
-
public:
+
+ // prediction hints (totally hacky)
+ //int predictions;
+ //string hint1;
+ //string hint2;
+ //string hint3;
+ //string hint4;
+ //string hint5;
+ //string hint6;
+ //string hint7;
+ //string hint8;
+ //string hint9;
+ //string hint10;
+
MClientRequest() {}
MClientRequest(int op, int client) : Message(MSG_CLIENT_REQUEST) {
memset(&st, 0, sizeof(st));
void set_iarg2(int i) { st.iarg2 = i; }
void set_targ(time_t& t) { st.targ = t; }
void set_targ2(time_t& t) { st.targ2 = t; }
- void set_sarg(string& arg) { this->sarg = arg; }
+ void set_sarg(string& arg) { this->sarg = arg; }
void set_sarg(const char *arg) { this->sarg = arg; }
void set_sarg2(string& arg) { this->sarg2 = arg; }
void set_sizearg(size_t s) { st.sizearg = s; }
void set_client_inst(const entity_inst_t& i) { st.client_inst = i; }
const entity_inst_t& get_client_inst() { return st.client_inst; }
+ //void set_num_hints(int hints) { st.num_hints = hints; }
+ //int get_num_hints() { return st.num_hints; }
+
int get_client() { return st.client; }
long get_tid() { return st.tid; }
int get_op() { return st.op; }
path._decode(payload, off);
_decode(sarg, payload, off);
_decode(sarg2, payload, off);
+
+ //payload.copy(off, sizeof(sarg2), (char*)&sarg2);
+ //off += sizeof(sarg2);
+ //payload.copy(off, sizeof(int), (char*)&predictions);
+ //off += sizeof(int);
+ //_decode(hint1, payload, off);
+ //_decode(hint2, payload, off);
+ //_decode(hint3, payload, off);
+ //_decode(hint4, payload, off);
+ //_decode(hint5, payload, off);
+ //_decode(hint6, payload, off);
+ //_decode(hint7, payload, off);
+ //_decode(hint8, payload, off);
+ //_decode(hint9, payload, off);
+ //_decode(hint10, payload, off);
}
virtual void encode_payload() {
path._encode(payload);
_encode(sarg, payload);
_encode(sarg2, payload);
+ //bufferlist helper;
+ //helper.append(sarg2.c_str(), sarg2.size());
+ //payload.append(helper);
+
+ //payload.append((char*)&sarg2, sizeof(sarg2));
+ //payload.append((char*)&predictions, sizeof(int));
+ //_encode(hint1, payload);
+ //_encode(hint2, payload);
+ //_encode(hint3, payload);
+ //_encode(hint4, payload);
+ //_encode(hint5, payload);
+ //_encode(hint6, payload);
+ //_encode(hint7, payload);
+ //_encode(hint8, payload);
+ //_encode(hint9, payload);
+ //_encode(hint10, payload);
}
void print(ostream& out) {
}
}
- #if (! $seen_paths{$dirpath}) {
- # $seen_paths{$dirpath} = 1;
- # print "mkdir\n\/$dirpath\n493\n";
- #}
-
}
}
pub_key = esignPubKey(tempKey);
//pub_str_key = pubToString(pub_key);
memcpy(pub_str_key, pubToString(pub_key).c_str(), sizeof(pub_str_key));
+
// now throw away the private key
keyConvert = false;
}
myPrivKey = _fromStr_esignPrivKey(string(pk, ESIGNPRIVSIZE));
myPubKey = esignPubKey(myPrivKey);
- // FIXME.
- //assert(0);
}
void Monitor::init()
// der?
//myPrivKey = esignPrivKey("crypto/esig1536.dat");
//myPubKey = esignPubKey(myPrivKey);
-
- // assert(0);
+
}
// create
// waiting for boot!
// add the OSD instance to the map?
osdmap.osd_inst[from] = m->get_source_inst();
+
// adds the key to the map
- osdmap.osd_str_keys[from] = m->get_public_key();
+ //osdmap.osd_str_keys[from] = m->get_public_key();
if (osdmap.osd_inst.size() == osdmap.osds.size()) {
dout(7) << "osd_boot all osds booted." << endl;
list<int> new_out;
map<int,float> new_overload; // updated overload value
list<int> old_overload; // no longer overload
- map<int,string> added_osd_keys; // new public keys
- list<int> removed_osd_keys; // public keys to remove
+ //map<int,string> added_osd_keys; // new public keys
+ //list<int> removed_osd_keys; // public keys to remove
void encode(bufferlist& bl) {
bl.append((char*)&epoch, sizeof(epoch));
::_encode(new_in, bl);
::_encode(new_out, bl);
::_encode(new_overload, bl);
- ::_encode(added_osd_keys, bl);
- ::_encode(removed_osd_keys, bl);
+ //::_encode(added_osd_keys, bl);
+ //::_encode(removed_osd_keys, bl);
}
void decode(bufferlist& bl, int& off) {
bl.copy(off, sizeof(epoch), (char*)&epoch);
::_decode(new_in, bl, off);
::_decode(new_out, bl, off);
::_decode(new_overload, bl, off);
- ::_decode(added_osd_keys, bl, off);
- ::_decode(removed_osd_keys, bl, off);
+ //::_decode(added_osd_keys, bl, off);
+ //::_decode(removed_osd_keys, bl, off);
}
Incremental(epoch_t e=0) : epoch(e), mon_epoch(0) {}
set<int> out_osds; // list of unmapped disks
map<int,float> overload_osds;
map<int,entity_inst_t> osd_inst;
- map<int,string> osd_str_keys; //all public keys in str form
- map<int,esignPub> osd_keys; // all public key objects (cache)
+ //map<int,string> osd_str_keys; //all public keys in str form
+ //map<int,esignPub> osd_keys; // all public key objects (cache)
public:
Crush crush; // hierarchical map
const set<int>& get_down_osds() { return down_osds; }
const set<int>& get_out_osds() { return out_osds; }
const map<int,float>& get_overload_osds() { return overload_osds; }
+ /*
const map<int,string>& get_key_str_map() { return osd_str_keys; }
const map<int,esignPub>& get_key_map() { return osd_keys; }
const esignPub get_key(int client) {
return tempPub;
}
const string& get_str_keys(int client) { return osd_str_keys[client]; }
+ */
bool is_down(int osd) { return down_osds.count(osd); }
bool is_up(int osd) { return !is_down(osd); }
overload_osds.erase(*i);
}
// add the incremental keys to osd_keys
+ /*
for (map<int,string>::iterator i = inc.added_osd_keys.begin();
i != inc.added_osd_keys.end();
i++) {
assert(osd_str_keys.count(*i)); // sanity check
osd_str_keys.erase(*i);
}
+ */
}
// serialize, unserialize
_encode(out_osds, blist);
_encode(overload_osds, blist);
_encode(osd_inst, blist);
- _encode(osd_keys, blist);
+ //_encode(osd_keys, blist);
crush._encode(blist);
}
_decode(out_osds, blist, off);
_decode(overload_osds, blist, off);
_decode(osd_inst, blist, off);
- _decode(osd_keys, blist, off);
+ //_decode(osd_keys, blist, off);
crush._decode(blist, off);
}
projects / ceph.git / commitdiff