Changeset 13812

Timestamp:

06/13/08 15:35:17 (5 years ago)

Author:

kazuho

Message:

support paging, implement maxid cache

Files:

• lang/sql/mysql_timeline/trunk/timeline.cc (modified) (4 diffs)

lang/sql/mysql_timeline/trunk/timeline.cc

@@ +1 @@
+#include <algorithm>
 #include "mysql_direct_access.hpp"
 
-#define MAX_ROWS 20
+#ifndef MAX_ROWS
+# define MAX_ROWS 20
+#endif
+
+#ifndef CACHE_MAX_MESSAGE_IDS
+# define CACHE_MAX_MESSAGE_IDS 0
+#endif
+
+#ifndef MAX_MESSAGE_IDS_INITIAL_CAPACITY
+# define MAX_MESSAGE_IDS_INITIAL_CAPACITY 4096
+#endif
+
+using namespace std;
+
+/* per user cache for keeping max. message id
+ * i.e. max_message_ids[$uid] == select max(id) from message where user_id=$uid
+ * 
+ * should be fullfilled on server startup by calling:
+ *  SELECT set_max_message_id(user_id,max(id)) FROM message GROUP BY user_id
+ */
+class max_message_ids_t {
+  
+protected:
+  unsigned *vect;
+  size_t vect_capacity;
+  pthread_rwlock_t rwlock;
+  
+public:
+  max_message_ids_t()
+  : vect((unsigned*)malloc(sizeof(unsigned)
+                           * MAX_MESSAGE_IDS_INITIAL_CAPACITY)),
+    vect_capacity(MAX_MESSAGE_IDS_INITIAL_CAPACITY)
+  {
+    assert(vect != NULL);
+    fill(vect, vect + vect_capacity, 0);
+    pthread_rwlockattr_t attr;
+    pthread_rwlockattr_init(&attr);
+    // switch to writer-preferred lock on linux
+#ifdef PTHREAD_RWLOCK_WRITER_NONRECURSIVE_INITIALIZER_NP
+    pthread_rwlockattr_setkind_np(&attr,
+                                  PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
+#endif
+    pthread_rwlock_init(&rwlock, &attr);
+    pthread_rwlockattr_destroy(&attr);
+  }
+  ~max_message_ids_t() {
+    free(vect);
+    vect = NULL;
+    pthread_rwlock_destroy(&rwlock);
+  }
+  void lock_reader() {
+    pthread_rwlock_rdlock(&rwlock);
+  }
+  void unlock_reader() {
+    pthread_rwlock_unlock(&rwlock);
+  }
+  unsigned get(unsigned user_id) {
+    return user_id < vect_capacity ? vect[user_id] : 0;
+  }
+  void set(unsigned user_id, unsigned message_id) {
+    pthread_rwlock_rdlock(&rwlock);
+    if (user_id < vect_capacity) {
+      vect[user_id] = message_id;
+      pthread_rwlock_unlock(&rwlock);
+      return;
+    } else {
+      pthread_rwlock_unlock(&rwlock);
+      pthread_rwlock_wrlock(&rwlock);
+      if (vect_capacity <= user_id) {
+        size_t old_capacity = vect_capacity;
+        do {
+        vect_capacity *= 2;
+        } while (vect_capacity <= user_id);
+        vect = (unsigned*)realloc(vect, sizeof(unsigned) * vect_capacity);
+        assert(vect != NULL);
+        fill(vect + old_capacity, vect + vect_capacity, 0);
+      }
+      vect[user_id] = message_id;
+      pthread_rwlock_unlock(&rwlock);
+    }
+  }
+};
+
+static max_message_ids_t max_message_ids;
+
+
+template <typename Elem> struct sorted_list_t {
+  
+  Elem list[MAX_ROWS]; /* top MAX_ROWS, sorted in desc. order */
+  int cnt;
+  
+  sorted_list_t() : cnt(0) {}
+  bool test_add(const Elem& item) {
+    int keep;
+    for (keep = cnt - 1; keep >= 0; keep--) {
+      if (item <= list[keep]) {
+        break;
+      }
+    }
+    if (keep == MAX_ROWS - 1) {
+      return false;
+    }
+    if (cnt < MAX_ROWS) {
+      cnt++;
+    }
+    for (int i = cnt - 2; i > keep; i--) {
+      list[i + 1] = list[i];
+    }
+    list[keep + 1] = item;
+    return true;
+  }
+};
+
+struct uid_maxid_t {
+  unsigned user_id;
+  unsigned max_id;
+  uid_maxid_t() {}
+  uid_maxid_t(unsigned u, unsigned m) : user_id(u), max_id(m) {}
+};
+
+bool operator<=(const uid_maxid_t& x, const uid_maxid_t& y) {
+  return x.max_id <= y.max_id;
+}
 
 struct timeline_t : public mysql_direct_access<timeline_t> {
   
-  unsigned id_list[MAX_ROWS]; /* top MAX_ROWS, sorted in desc. order */
-  int id_cnt;
-  
-  timeline_t() : mysql_direct_access<timeline_t>(), id_cnt(0) {}
-  
   static int type_init(UDF_INIT *initid, UDF_ARGS *args, char *message) {
-    if (args->arg_count != 1) {
-      strcpy(message, "timeline(user_id)");
+    switch (args->arg_count) {
+    case 2:
+      args->arg_type[1] = INT_RESULT;
+      args->maybe_null[1] = 0;
+      // don't break
+    case 1:
+      args->arg_type[0] = INT_RESULT;
+      args->maybe_null[0] = 0;
+      break;
+    default:
+      strcpy(message, "timeline(user_id[,max_message_id])");
       return 1;
     }
-    args->arg_type[0] = INT_RESULT;
-    args->maybe_null[0] = 0;
     initid->maybe_null = 0;
     return 0;
@@ -28 +153 @@
   }
   
-  int do_task() {
-    
-    unsigned user_id = *(long long*)args->args[0];
-    
-    TABLE *message_tbl = get_table(0);
-    Field *message_id_fld = get_field(message_tbl, "id");
+  TABLE *message_tbl;
+  Field *message_id_fld;
+  Field *message_user_id_fld;
+  KEY *message_key;
+  TABLE *follower_tbl;
+  Field *follower_user_id_fld;
+  Field *follower_follower_id_fld;
+  KEY *follower_key;
+  sorted_list_t<unsigned> message_ids;
+  
+  void setup() {
+    message_tbl = get_table(0);
+    message_id_fld = get_field(message_tbl, "id");
     assert(message_id_fld != NULL);
-    Field *message_user_id_fld = get_field(message_tbl, "user_id");
+    message_user_id_fld = get_field(message_tbl, "user_id");
     assert(message_user_id_fld != NULL);
     message_tbl->clear_column_bitmaps();
     bitmap_set_bit(message_tbl->read_set, message_id_fld->field_index);
     bitmap_set_bit(message_tbl->read_set, message_user_id_fld->field_index);
-    KEY *message_key = index_init(message_tbl, "user_id_id", true);
+    message_key = index_init(message_tbl, "user_id_id", true);
     assert(message_key != NULL);
-    uchar* message_keybuff = (uchar*)alloca(message_key->key_length);    
-    memset(message_keybuff, 0, message_key->key_length);
-    
-    TABLE *follower_tbl = get_table(1);
-    Field *follower_user_id_fld = get_field(follower_tbl, "user_id");
+    
+    follower_tbl = get_table(1);
+    follower_user_id_fld = get_field(follower_tbl, "user_id");
     assert(follower_user_id_fld != NULL);
-    Field *follower_follower_id_fld = get_field(follower_tbl, "follower_id");
+    follower_follower_id_fld = get_field(follower_tbl, "follower_id");
     assert(follower_follower_id_fld != NULL);
     follower_tbl->use_all_columns();
-    KEY *follower_key = index_init(follower_tbl, "PRIMARY", false);
+    follower_key = index_init(follower_tbl, "PRIMARY", false);
     assert(follower_key != NULL);
-    uchar* follower_keybuff = (uchar*)alloca(follower_key->key_length);
-    memset(follower_keybuff, 0, follower_key->key_length);
+  }
+  
+  void per_user_search(unsigned follower_id, unsigned max_id) {
+    uchar *keybuff = (uchar*)alloca(message_key->key_length);
+    
+    if (max_id == 0) {
+      int4store(keybuff, follower_id);
+      if (message_tbl->file->index_read_map(message_tbl->record[0], keybuff, 1,
+                                            HA_READ_PREFIX_LAST)
+          != 0) {
+        return;
+      }
+    } else {
+      int4store(keybuff, follower_id);
+      int4store(keybuff + 4, max_id);
+      if (message_tbl->file->index_read_map(message_tbl->record[0], keybuff, 3,
+                                            HA_READ_BEFORE_KEY)
+          != 0 ||
+          message_user_id_fld->val_int() != follower_id) {
+        return;
+      }
+    }
+    do {
+      if (! message_ids.test_add(message_id_fld->val_int())) {
+        break;
+      }
+    } while (message_tbl->file->index_prev(message_tbl->record[0]) == 0
+             && message_user_id_fld->val_int() == follower_id);
+  }
+  
+  int do_task() {
+    
+    unsigned user_id = *(long long*)args->args[0];
+    unsigned max_id = args->arg_count >= 2 ? *(long long*)args->args[1] : 0;
+    
+    setup();
     
     /* main */
-    int4store(follower_keybuff, user_id);
-    if (follower_tbl->file->index_read_map(follower_tbl->record[0],
-                                           follower_keybuff, 1, HA_READ_PREFIX)
+    uchar *keybuff = (uchar*)alloca(follower_key->key_length);
+    memset(keybuff, 0, follower_key->key_length);
+    int4store(keybuff, user_id);
+    if (follower_tbl->file->index_read_map(follower_tbl->record[0], keybuff,
+                                           1, HA_READ_PREFIX)
         == 0) {
-      do {
-        unsigned follower_id = follower_follower_id_fld->val_int();
-        int4store(message_keybuff, follower_id);
-        if (message_tbl->file->index_read_map(message_tbl->record[0],
-                                              message_keybuff, 1,
-                                              HA_READ_PREFIX_LAST)
-            == 0) {
-          do {
-            if (! test_add_id(message_id_fld->val_int())) {
-              break;
-            }
-          } while (message_tbl->file->index_prev(message_tbl->record[0]) == 0
-                   && message_user_id_fld->val_int() == follower_id);
+      if (CACHE_MAX_MESSAGE_IDS && max_id == 0) {
+        sorted_list_t<uid_maxid_t> uid_maxid_list;
+        max_message_ids.lock_reader();
+        do {
+          unsigned fid = follower_follower_id_fld->val_int();
+          uid_maxid_list.test_add(uid_maxid_t(fid, max_message_ids.get(fid)));
+        } while (follower_tbl->file->index_next(follower_tbl->record[0]) == 0
+                 && follower_user_id_fld->val_int() == user_id);
+        max_message_ids.unlock_reader();
+        for (int i = 0;
+             i < uid_maxid_list.cnt && uid_maxid_list.list[i].max_id != 0;
+             i++) {
+          per_user_search(uid_maxid_list.list[i].user_id, 0);
         }
-      } while (follower_tbl->file->index_next(follower_tbl->record[0]) == 0
-               && follower_user_id_fld->val_int() == user_id);
+      } else {
+        do {
+          unsigned fid = follower_follower_id_fld->val_int();
+          per_user_search(fid, max_id);
+        } while (follower_tbl->file->index_next(follower_tbl->record[0]) == 0
+                 && follower_user_id_fld->val_int() == user_id);
+      }
     }
     follower_tbl->file->ha_index_end();
@@ -87 +258 @@
     out_tt->use_all_columns();
     out_tt->file->ha_delete_all_rows();
-    out_tt->file->ha_start_bulk_insert(id_cnt);
+    out_tt->file->ha_start_bulk_insert(message_ids.cnt);
     memset(out_tt->record[0], 0, out_tt->s->null_bytes);
-    for (int i = 0; i < id_cnt; i++) {
-      out_tt_id_fld->store(id_list[i], true);
+    for (int i = 0; i < message_ids.cnt; i++) {
+      out_tt_id_fld->store(message_ids.list[i], true);
       out_tt->file->ha_write_row(out_tt->record[0]);
     }
@@ -97 +268 @@
     return 0;
   }
-
-  bool test_add_id(unsigned id) {
-    int keep;
-    for (keep = id_cnt - 1; keep >= 0; keep--) {
-      if (id <= id_list[keep]) {
-        break;
-      }
-    }
-    if (keep == MAX_ROWS - 1) {
-      return false;
-    }
-    if (id_cnt < MAX_ROWS) {
-      id_cnt++;
-    }
-    for (int i = id_cnt - 2; i > keep; i--) {
-      id_list[i + 1] = id_list[i];
-    }
-    id_list[keep + 1] = id;
-    return true;
-  }
 };
 
 MYSQL_DIRECT_ACCESS_DECLARE_AS_INT_UDF(timeline_t, timeline);
+
+extern "C" my_bool timeline_set_maxid_init(UDF_INIT *initid, UDF_ARGS *args,
+                                       char *message)
+{
+  if (args->arg_count != 2) {
+    strcpy(message, "usage: timeline_set_maxid(user_id,max_id)");
+    return 1;
+  }
+  args->arg_type[0] = INT_RESULT;
+  args->maybe_null[0] = 0;
+  args->arg_type[1] = INT_RESULT;
+  args->maybe_null[1] = 0;
+  initid->maybe_null = 0;
+  return 0;
+}
+
+extern "C" void timeline_set_maxid_deinit(UDF_INIT *initid)
+{
+}
+
+extern "C" long long timeline_set_maxid(UDF_INIT *initid, UDF_ARGS *args,
+                                        char *is_null, char *error)
+{
+  max_message_ids.set(*(long long*)args->args[0], *(long long*)args->args[1]);
+  return 0;
+}