#93 storage interface

- added save_mail

src/librmb/rados-mail-object.h

@@ -35,20 +35,21 @@ class RadosMailObject {
   void set_state(const string& _state) { this->state = _state; }
   void set_version(const string& _version) { this->version = _version; }
   void set_guid(const uint8_t* guid);
-  void set_object_size(const uint64_t& _size) { object_size = _size; }
+  void set_mail_size(const uint64_t& _size) { object_size = _size; }
   void set_mail_buffer(char* _mail_buffer) { this->mail_buffer = _mail_buffer; }
   void set_active_op(bool _active) { this->active_op = _active; }
   void set_rados_save_date(const time_t& _save_date) { this->save_date_rados = _save_date; }
 
   const string get_oid() { return this->oid; }
   const string get_version() { return this->version; }
-  const uint64_t& get_object_size() { return this->object_size; }
+  const uint64_t& get_mail_size() { return this->object_size; }
 
   time_t* get_rados_save_date() { return &this->save_date_rados; }
   uint8_t* get_guid_ref() { return guid; }
   char* get_mail_buffer() { return this->mail_buffer; }
   map<string, ceph::bufferlist>* get_metadata() { return &this->attrset; }
-
+  void set_mail_buffer_content_ptr(const void* start) { mail_buffer_start = start; }
+  const void* get_mail_buffer_content_ptr() { return mail_buffer_start; }
   map<AioCompletion*, ObjectWriteOperation*>* get_completion_op_map() { return &completion_op; }
   string get_metadata(rbox_metadata_key key) {
     string str_key(1, static_cast<char>(key));
@@ -64,6 +65,7 @@ class RadosMailObject {
 
   bool has_active_op() { return active_op; }
   string to_string(const string& padding);
+  void add_metadata(RadosMetadata& metadata) { attrset[metadata.key] = metadata.bl; }
 
  private:
   string oid;
@@ -74,6 +76,7 @@ class RadosMailObject {
   uint8_t guid[GUID_128_SIZE];
   uint64_t object_size;  // byte
   map<AioCompletion*, ObjectWriteOperation*> completion_op;
+  const void* mail_buffer_start;
 
   bool active_op;
   // used as pointer to a buffer_t (to avoid using dovecot datatypes in library)

src/librmb/rados-storage-impl.cpp

@@ -26,7 +26,7 @@ const char *RadosStorageImpl::CFG_OSD_MAX_WRITE_SIZE = "osd_max_write_size";
 
 RadosStorageImpl::RadosStorageImpl(RadosCluster *_cluster) {
   cluster = _cluster;
-  max_write_size = 0;
+  max_write_size = 10;
 }
 
 RadosStorageImpl::~RadosStorageImpl() {}
@@ -178,6 +178,25 @@ bool RadosStorageImpl::wait_for_write_operations_complete(
   return failed;
 }
 
+bool RadosStorageImpl::wait_for_rados_operations(const std::vector<librmb::RadosMailObject *> &object_list) {
+  bool ctx_failed = false;
+  // wait for all writes to finish!
+  // imaptest shows it's possible that begin -> continue -> finish cycle is invoked several times before
+  // rbox_transaction_save_commit_pre is called.
+  for (std::vector<librmb::RadosMailObject *>::const_iterator it_cur_obj = object_list.begin();
+       it_cur_obj != object_list.end(); ++it_cur_obj) {
+    // if we come from copy mail, there is no operation to wait for.
+    if ((*it_cur_obj)->has_active_op()) {
+      bool op_failed = wait_for_write_operations_complete((*it_cur_obj)->get_completion_op_map());
+
+      ctx_failed = ctx_failed ? ctx_failed : op_failed;
+      (*it_cur_obj)->get_completion_op_map()->clear();
+      (*it_cur_obj)->set_active_op(false);
+    }
+  }
+  return ctx_failed;
+}
+
 bool RadosStorageImpl::update_metadata(std::string oid, std::list<RadosMetadata> &to_update) {
   librados::ObjectWriteOperation write_op;
   librados::AioCompletion *completion = librados::Rados::aio_create_completion();
@@ -276,5 +295,32 @@ bool RadosStorageImpl::copy(std::string &src_oid, const char *src_ns, std::strin
   return ret == 0;
 }
 
+// if save_async = true, don't forget to call wait_for_rados_operations e.g. wait_for_write_operations_complete
+// to wait for completion and free resources.
+bool RadosStorageImpl::save_mail(RadosMailObject *mail, bool &save_async) {
+  // delete write_op_xattr is called after operation completes (wait_for_rados_operations)
+  librados::ObjectWriteOperation *write_op_xattr = new librados::ObjectWriteOperation();
+  int ret = -1;
+  // set metadata
+  for (std::map<std::string, librados::bufferlist>::iterator it = mail->get_metadata()->begin();
+       it != mail->get_metadata()->end(); ++it) {
+    write_op_xattr->setxattr(it->first.c_str(), it->second);
+  }
+  write_op_xattr->mtime(mail->get_rados_save_date());
+
+  ret = split_buffer_and_exec_op(reinterpret_cast<const char *>(mail->get_mail_buffer_content_ptr()),
+                                 mail->get_mail_size(), mail, write_op_xattr, get_max_write_size_bytes());
+  mail->set_active_op(true);
+  if (!save_async) {
+    std::vector<librmb::RadosMailObject *> objects;
+    objects.push_back(mail);
+    return wait_for_rados_operations(objects);
+  }
+  if (ret != 0) {
+    write_op_xattr->remove();
+    delete write_op_xattr;
+    mail->set_active_op(false);
+  }
 
-
+  return ret == 0;
+}

src/librmb/rados-storage-impl.h

@@ -53,13 +53,17 @@ class RadosStorageImpl : public RadosStorage {
   bool wait_for_write_operations_complete(
       std::map<librados::AioCompletion *, librados::ObjectWriteOperation *> *completion_op_map);
 
+  bool wait_for_rados_operations(const std::vector<librmb::RadosMailObject *> &object_list);
+
   int read_mail(librados::bufferlist *buffer, const std::string &oid);
   bool update_metadata(std::string oid, std::list<RadosMetadata> &to_update);
   bool move(std::string &src_oid, const char *src_ns, std::string &dest_oid, const char *dest_ns,
             std::list<RadosMetadata> &to_update, bool delete_source);
   bool copy(std::string &src_oid, const char *src_ns, std::string &dest_oid, const char *dest_ns,
             std::list<RadosMetadata> &to_update);
 
+  bool save_mail(RadosMailObject *mail, bool &save_async);
+
  private:
   RadosCluster *cluster;
   int max_write_size;

src/librmb/rados-storage.h

@@ -47,12 +47,15 @@ class RadosStorage {
   virtual int open_connection(const std::string &poolname, const std::string &ns) = 0;
   virtual bool wait_for_write_operations_complete(
       std::map<librados::AioCompletion*, librados::ObjectWriteOperation*>* completion_op_map) = 0;
+  virtual bool wait_for_rados_operations(const std::vector<librmb::RadosMailObject *> &object_list) = 0;
+
   virtual int read_mail(librados::bufferlist *buffer, const std::string &oid) = 0;
   virtual bool update_metadata(std::string oid, std::list<RadosMetadata> &to_update) = 0;
   virtual bool move(std::string &src_oid, const char *src_ns, std::string &dest_oid, const char *dest_ns,
                     std::list<RadosMetadata> &to_update, bool delete_source) = 0;
   virtual bool copy(std::string &src_oid, const char *src_ns, std::string &dest_oid, const char *dest_ns,
                     std::list<RadosMetadata> &to_update) = 0;
+  virtual bool save_mail(RadosMailObject *mail, bool &save_async) = 0;
 };
 
 }  // namespace librmb

src/librmb/tools/rmb/mailbox_tools.cpp

@@ -89,7 +89,7 @@ int MailboxTools::save_mail(librmb::RadosMailObject* mail_obj) {
   if (!myfile.is_open()) {
     return -1;
   }
-  myfile.write(mail_obj->get_mail_buffer(), mail_obj->get_object_size());
+  myfile.write(mail_obj->get_mail_buffer(), mail_obj->get_mail_size());
   myfile.close();
   return 0;
 }

src/librmb/tools/rmb/rmb.cpp

@@ -170,12 +170,12 @@ static void query_mail_storage(std::vector<librmb::RadosMailObject *> *mail_obje
     }
     if (mailbox.count(mailbox_guid) > 0) {
       mailbox[mailbox_guid]->add_mail((*it));
-      mailbox[mailbox_guid]->add_to_mailbox_size((*it)->get_object_size());
+      mailbox[mailbox_guid]->add_to_mailbox_size((*it)->get_mail_size());
     } else {
       mailbox[mailbox_guid] = new librmb::RadosMailBox(mailbox_guid, 1, mailbox_orig_name);
       mailbox[mailbox_guid]->set_xattr_filter(parser);
       mailbox[mailbox_guid]->add_mail((*it));
-      mailbox[mailbox_guid]->add_to_mailbox_size((*it)->get_object_size());
+      mailbox[mailbox_guid]->add_to_mailbox_size((*it)->get_mail_size());
     }
   }
   std::cout << "mailbox_count: " << mailbox.size() << std::endl;
@@ -207,7 +207,7 @@ static void query_mail_storage(std::vector<librmb::RadosMailObject *> *mail_obje
           char *mail_buffer = new char[size_r + 1];
           (*it_mail)->set_mail_buffer(mail_buffer);
 
-          (*it_mail)->set_object_size(size_r);
+          (*it_mail)->set_mail_size(size_r);
           int read = storage->read_mail(&buffer, oid);
           if (read > 0) {
             memcpy(mail_buffer, buffer.to_str().c_str(), read + 1);
@@ -285,7 +285,7 @@ static void load_objects(librmb::RadosStorageImpl &storage, std::vector<librmb::
     time_t save_date_rados = 0;
     storage.stat_mail(iter->get_oid(), &object_size, &save_date_rados);
 
-    mail->set_object_size(object_size);
+    mail->set_mail_size(object_size);
     mail->set_rados_save_date(save_date_rados);
     ++iter;
     mail_objects.push_back(mail);

src/storage-rbox/rbox-mail.cpp

@@ -90,7 +90,7 @@ int rbox_get_index_record(struct mail *_mail) {
     rmail->mail_object->get_metadata()->clear();
   }
   uint64_t obj_size = -1;
-  rmail->mail_object->set_object_size(obj_size);
+  rmail->mail_object->set_mail_size(obj_size);
   FUNC_END();
   return 0;
 }