#include <obd_class.h>
#include <lustre_log.h>
#include <libcfs/list.h>
+#include "llog_internal.h"
/* Allocate a new log or catalog handle */
struct llog_handle *llog_alloc_handle(void)
int rc = 0;
ENTRY;
- CDEBUG(D_HA, "canceling %d in log "LPX64"\n",
+ CDEBUG(D_RPCTRACE, "canceling %d in log "LPX64"\n",
index, loghandle->lgh_id.lgl_oid);
if (index == 0) {
}
if (!ext2_clear_bit(index, llh->llh_bitmap)) {
- CDEBUG(D_HA, "catalog index %u already clear?\n", index);
+ CDEBUG(D_RPCTRACE, "catalog index %u already clear?\n", index);
RETURN(-EINVAL);
}
llh->llh_hdr.lrh_type = LLOG_HDR_MAGIC;
llh->llh_hdr.lrh_len = llh->llh_tail.lrt_len = LLOG_CHUNK_SIZE;
llh->llh_hdr.lrh_index = llh->llh_tail.lrt_index = 0;
- llh->llh_timestamp = CURRENT_SECONDS;
+ llh->llh_timestamp = cfs_time_current_sec();
if (uuid)
memcpy(&llh->llh_tgtuuid, uuid, sizeof(llh->llh_tgtuuid));
llh->llh_bitmap_offset = offsetof(typeof(*llh),llh_bitmap);
}
EXPORT_SYMBOL(llog_close);
-int llog_process(struct llog_handle *loghandle, llog_cb_t cb,
- void *data, void *catdata)
+static int llog_process_thread(void *arg)
{
- struct llog_log_hdr *llh = loghandle->lgh_hdr;
- struct llog_process_cat_data *cd = catdata;
- char *buf;
- __u64 cur_offset = LLOG_CHUNK_SIZE, last_offset;
- int rc = 0, index = 1, last_index;
- int saved_index = 0, last_called_index = 0;
- ENTRY;
+ struct llog_process_info *lpi = (struct llog_process_info *)arg;
+ struct llog_handle *loghandle = lpi->lpi_loghandle;
+ struct llog_log_hdr *llh = loghandle->lgh_hdr;
+ struct llog_process_cat_data *cd = lpi->lpi_catdata;
+ char *buf;
+ __u64 cur_offset = LLOG_CHUNK_SIZE;
+ __u64 last_offset;
+ int rc = 0, index = 1, last_index;
+ int saved_index = 0, last_called_index = 0;
LASSERT(llh);
OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
- if (!buf)
- RETURN(-ENOMEM);
+ if (!buf) {
+ lpi->lpi_rc = -ENOMEM;
+#ifdef __KERNEL__
+ complete(&lpi->lpi_completion);
+#endif
+ return 0;
+ }
+
+ cfs_daemonize_ctxt("llog_process_thread");
if (cd != NULL) {
last_called_index = cd->first_idx;
index, last_index);
/* get the buf with our target record; avoid old garbage */
+ memset(buf, 0, LLOG_CHUNK_SIZE);
last_offset = cur_offset;
rc = llog_next_block(loghandle, &saved_index, index,
&cur_offset, buf, LLOG_CHUNK_SIZE);
CDEBUG(D_OTHER, "after swabbing, type=%#x idx=%d\n",
rec->lrh_type, rec->lrh_index);
-
+
if (rec->lrh_index == 0)
GOTO(out, 0); /* no more records */
continue;
}
- CDEBUG(D_OTHER,
+ CDEBUG(D_OTHER,
"lrh_index: %d lrh_len: %d (%d remains)\n",
rec->lrh_index, rec->lrh_len,
(int)(buf + LLOG_CHUNK_SIZE - (char *)rec));
loghandle->lgh_cur_idx = rec->lrh_index;
loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
- last_offset;
+ last_offset;
/* if set, process the callback on this record */
if (ext2_test_bit(index, llh->llh_bitmap)) {
- rc = cb(loghandle, rec, data);
+ rc = lpi->lpi_cb(loghandle, rec,
+ lpi->lpi_cbdata);
last_called_index = index;
if (rc == LLOG_PROC_BREAK) {
CDEBUG(D_HA, "recovery from log: "LPX64
loghandle->lgh_id.lgl_ogen);
GOTO(out, rc);
} else if (rc == LLOG_DEL_RECORD) {
- llog_cancel_rec(loghandle, rec->lrh_index);
+ llog_cancel_rec(loghandle,
+ rec->lrh_index);
rc = 0;
}
if (rc)
cd->last_idx = last_called_index;
if (buf)
OBD_FREE(buf, LLOG_CHUNK_SIZE);
+ lpi->lpi_rc = rc;
+#ifdef __KERNEL__
+ complete(&lpi->lpi_completion);
+#endif
+ return 0;
+}
+
+int llog_process(struct llog_handle *loghandle, llog_cb_t cb,
+ void *data, void *catdata)
+{
+ struct llog_process_info *lpi;
+ int rc;
+ ENTRY;
+
+ OBD_ALLOC_PTR(lpi);
+ if (lpi == NULL) {
+ CERROR("cannot alloc pointer\n");
+ RETURN(-ENOMEM);
+ }
+ lpi->lpi_loghandle = loghandle;
+ lpi->lpi_cb = cb;
+ lpi->lpi_cbdata = data;
+ lpi->lpi_catdata = catdata;
+
+#ifdef __KERNEL__
+ init_completion(&lpi->lpi_completion);
+ rc = cfs_kernel_thread(llog_process_thread, lpi, CLONE_VM | CLONE_FILES);
+ if (rc < 0) {
+ CERROR("cannot start thread: %d\n", rc);
+ OBD_FREE_PTR(lpi);
+ RETURN(rc);
+ }
+ wait_for_completion(&lpi->lpi_completion);
+#else
+ llog_process_thread(lpi);
+#endif
+ rc = lpi->lpi_rc;
+ OBD_FREE_PTR(lpi);
RETURN(rc);
}
EXPORT_SYMBOL(llog_process);
rec = buf;
idx = le32_to_cpu(rec->lrh_index);
if (idx < index)
- CDEBUG(D_HA, "index %u : idx %u\n", index, idx);
+ CDEBUG(D_RPCTRACE, "index %u : idx %u\n", index, idx);
while (idx < index) {
rec = ((void *)rec + le32_to_cpu(rec->lrh_len));
idx ++;