<commit_before>/* $Id$ */

#define PSC_SUBSYS PSS_JOURNAL
#include "psc_util/subsys.h"

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include <sys/types.h>

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <string.h>
#include <unistd.h>

#include "psc_util/alloc.h"
#include "psc_util/atomic.h"
#include "psc_util/journal.h"
#include "psc_util/lock.h"
#include "psc_ds/dynarray.h"

static struct psc_journal_xidhndl *
pjournal_xidhndl_new(struct psc_journal *pj)
{
	struct psc_journal_xidhndl *xh;

	xh = PSCALLOC(sizeof(*xh));

	xh->pjx_pj = pj;

	psc_warnx("xh=%p xh->pjx_pj=%p", xh, xh->pjx_pj);

	xh->pjx_tailslot = PJX_SLOT_ANY;
	INIT_PSCLIST_ENTRY(&xh->pjx_lentry);
	LOCK_INIT(&xh->pjx_lock);
	return (xh);
}

/*
 * pjournal_nextxid - obtain an unused journal transaction ID.
 * @pj: the journal.
 * Returns: new, unused transaction ID.
 */
struct psc_journal_xidhndl *
pjournal_nextxid(struct psc_journal *pj)
{
	struct psc_journal_xidhndl *xh;

	xh = pjournal_xidhndl_new(pj);

	PJ_LOCK(pj);
	do {
		xh->pjx_xid = ++pj->pj_nextxid;
	} while (xh->pjx_xid == PJE_XID_NONE);
	PJ_ULOCK(pj);

	psc_assert(xh->pjx_pj == pj);

	return (xh);
}

/*
 * _pjournal_logwrite - store a new entry in a journal.
 * @pj: the journal.
 * @slot: position location in journal to write.
 * @type: the application-specific log entry type.
 * @xid: transaction ID of entry.
 * @data: the journal entry contents to store.
 * Returns: 0 on success, -1 on error.
 */
__static int
pjournal_logwrite_internal(struct psc_journal *pj, struct psc_journal_xidhndl *xh,
		   	    uint32_t slot, int type, void *data, size_t size)
{
	struct psc_journal_enthdr *pje;
	int rc = 0, len;

	psc_assert(slot < pj->pj_hdr->pjh_nents);
	psc_assert(size <= PJ_PJESZ(pj));

#ifdef PJE_DYN_BUFFER
	pje = psc_alloc(PJ_PJESZ(pj), PAF_PAGEALIGN | PAF_LOCK);
#else
	PJ_LOCK(pj);
	while (!(len = dynarray_len(&pj->pj_bufs))) {
		psc_waitq_wait(&pj->pj_waitq, &pj->pj_lock);
		PJ_LOCK(pj);
	}
	pje = dynarray_getpos(&pj->pj_bufs, len-1);
	psc_assert(pje);
	dynarray_remove(&pj->pj_bufs, pje);
	psc_notify("got pje=%p", pje);
	PJ_ULOCK(pj);
#endif
	if (data)
		memcpy(pje->pje_data, data, size);

	pje->pje_magic = PJE_MAGIC;
	pje->pje_type = type;
	pje->pje_xid = xh->pjx_xid;

	if (!(type & PJET_XEND))
		pje->pje_sid = atomic_inc_return(&xh->pjx_sid);
	else
		pje->pje_sid = atomic_read(&xh->pjx_sid);

#ifdef NOT_READY
	/* commit the log on disk before we can return */
	rc = pwrite(pj->pj_fd, pje, pj->pj_hdr->pjh_entsz, 
		   (off_t)(pj->pj_hdr->pjh_start_off + (slot * pj->pj_hdr->pjh_entsz)));
	if (rc != -1 && rc != pj->pj_hdr->pjh_entsz)
		rc = -EAGAIN;
#endif

#ifdef PJE_DYN_BUFFER
	psc_freel(pje, PJ_PJESZ(pj));
#else
	PJ_LOCK(pj);
	dynarray_add(&pj->pj_bufs, pje);
	psc_waitq_wakeall(&pj->pj_waitq);
	PJ_ULOCK(pj);
#endif
	return (rc);
}

/*
 * _pjournal_logwritex - store a new entry in a journal transaction.
 * @pj: the journal.
 * @type: the application-specific log entry type.
 * @xid: transaction ID.
 * @data: the journal entry contents to store.
 * Returns: 0 on success, -1 on error.
 */
int
pjournal_logwrite(struct psc_journal_xidhndl *xh, int type, void *data,
		  size_t size)
{
	struct psc_journal_xidhndl	*t;
	int				 rc;
	struct psc_journal		*pj;
	uint32_t			 slot;
	uint32_t			 tail_slot;

	if (type & PJET_NODATA)
		psc_assert(!data);

	tail_slot = 0;
	pj = xh->pjx_pj;

	psc_assert(!(type & PJET_CORRUPT));
	psc_assert(!(type & PJET_XSTARTED));
	psc_assert(!(type & PJET_XEND));

	psc_assert(!(xh->pjx_flags & PJX_XCLOSED));

 retry:
	/*
	 * Make sure that the next slot to be written does not have a pending transaction.
	 * Since we add a new transaction at the tail of the pending transaction list, we
	 * only need to check the head of the list to find out the oldest pending transaction.
	 */
	PJ_LOCK(pj);
	slot = pj->pj_nextwrite;
	t = psclist_first_entry(&pj->pj_pndgxids, struct psc_journal_xidhndl, pjx_lentry);
	if (t) {
		if (t->pjx_tailslot == slot) {
			psc_warnx("pj(%p) blocking on slot(%d) "
				  "availability - owned by xid (%p)", 
				  pj, slot, t);
			psc_waitq_wait(&pj->pj_waitq, &pj->pj_lock);
			goto retry;
		}
		tail_slot = t->pjx_tailslot;
	}

	psc_info("pj(%p) tail@slot(%d) my@slot(%d) xh_flags(%o)",
		 pj, tail_slot, slot, xh->pjx_flags);

	if (xh->pjx_flags & PJET_XEND) {
		if (xh->pjx_flags & PJET_XSTARTED) {
			xh->pjx_flags |= PJET_XCLOSED;
			psclist_del(&xh->pjx_lentry);
			if (xh->pjx_tailslot == pj->pj_nextwrite) {
				/* We are the tail so unblock the journal.
				 */
				psc_warnx("pj(%p) unblocking slot(%d) - "
					  "owned by xid (%p)", 
					  pj, slot, xh);
				psc_waitq_wakeall(&pj->pj_waitq);
			}
		}
	} else if (!(xh->pjx_flags & PJET_XSTARTED)) {
		/* Multi-step operation, mark the slot id here
		 *  so that the tail of the journal can be found
		 *  and that overwriting pending xids may be
		 *  prevented.
		 * Note:  self-contained ops (PJET_XEND)
		 *        cannot end up here.
		 */
		psc_assert(!(xh->pjx_flags & PJET_XEND));

		xh->pjx_tailslot = slot;
		psclist_xadd_tail(&xh->pjx_lentry, &pj->pj_pndgxids);
		xh->pjx_flags |= PJET_XSTARTED;
	}<commit_msg>wake up threads only after the transaction is committed ON DISK<commit_after>/* $Id$ */

#define PSC_SUBSYS PSS_JOURNAL
#include "psc_util/subsys.h"

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include <sys/types.h>

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <string.h>
#include <unistd.h>

#include "psc_util/alloc.h"
#include "psc_util/atomic.h"
#include "psc_util/journal.h"
#include "psc_util/lock.h"
#include "psc_ds/dynarray.h"

static struct psc_journal_xidhndl *
pjournal_xidhndl_new(struct psc_journal *pj)
{
	struct psc_journal_xidhndl *xh;

	xh = PSCALLOC(sizeof(*xh));

	xh->pjx_pj = pj;

	psc_warnx("xh=%p xh->pjx_pj=%p", xh, xh->pjx_pj);

	xh->pjx_tailslot = PJX_SLOT_ANY;
	INIT_PSCLIST_ENTRY(&xh->pjx_lentry);
	LOCK_INIT(&xh->pjx_lock);
	return (xh);
}

/*
 * pjournal_nextxid - obtain an unused journal transaction ID.
 * @pj: the journal.
 * Returns: new, unused transaction ID.
 */
struct psc_journal_xidhndl *
pjournal_nextxid(struct psc_journal *pj)
{
	struct psc_journal_xidhndl *xh;

	xh = pjournal_xidhndl_new(pj);

	PJ_LOCK(pj);
	do {
		xh->pjx_xid = ++pj->pj_nextxid;
	} while (xh->pjx_xid == PJE_XID_NONE);
	PJ_ULOCK(pj);

	psc_assert(xh->pjx_pj == pj);

	return (xh);
}

/*
 * _pjournal_logwrite - store a new entry in a journal.
 * @pj: the journal.
 * @slot: position location in journal to write.
 * @type: the application-specific log entry type.
 * @xid: transaction ID of entry.
 * @data: the journal entry contents to store.
 * Returns: 0 on success, -1 on error.
 */
__static int
pjournal_logwrite_internal(struct psc_journal *pj, struct psc_journal_xidhndl *xh,
		   	    uint32_t slot, int type, void *data, size_t size)
{
	struct psc_journal_enthdr *pje;
	int rc = 0, len;

	psc_assert(slot < pj->pj_hdr->pjh_nents);
	psc_assert(size <= PJ_PJESZ(pj));

#ifdef PJE_DYN_BUFFER
	pje = psc_alloc(PJ_PJESZ(pj), PAF_PAGEALIGN | PAF_LOCK);
#else
	PJ_LOCK(pj);
	while (!(len = dynarray_len(&pj->pj_bufs))) {
		psc_waitq_wait(&pj->pj_waitq, &pj->pj_lock);
		PJ_LOCK(pj);
	}
	pje = dynarray_getpos(&pj->pj_bufs, len-1);
	psc_assert(pje);
	dynarray_remove(&pj->pj_bufs, pje);
	psc_notify("got pje=%p", pje);
	PJ_ULOCK(pj);
#endif
	if (data)
		memcpy(pje->pje_data, data, size);

	pje->pje_magic = PJE_MAGIC;
	pje->pje_type = type;
	pje->pje_xid = xh->pjx_xid;

	if (!(type & PJET_XEND))
		pje->pje_sid = atomic_inc_return(&xh->pjx_sid);
	else
		pje->pje_sid = atomic_read(&xh->pjx_sid);

#ifdef NOT_READY
	/* commit the log on disk before we can return */
	rc = pwrite(pj->pj_fd, pje, pj->pj_hdr->pjh_entsz, 
		   (off_t)(pj->pj_hdr->pjh_start_off + (slot * pj->pj_hdr->pjh_entsz)));
	if (rc != -1 && rc != pj->pj_hdr->pjh_entsz)
		rc = -EAGAIN;
#endif

#ifdef PJE_DYN_BUFFER
	psc_freel(pje, PJ_PJESZ(pj));
#else
	PJ_LOCK(pj);
	dynarray_add(&pj->pj_bufs, pje);
	psc_waitq_wakeall(&pj->pj_waitq);
	PJ_ULOCK(pj);
#endif
	if (xh->pjx_flags & PJX_XCLOSED && xh->pjx_tailslot == pj->pj_nextwrite) {
		/* We are the tail so unblock the journal.  */
		psc_warnx("pj(%p) unblocking slot(%d) - "
			  "owned by xid (%p)", 
			  pj, slot, xh);
		psc_waitq_wakeall(&pj->pj_waitq);
	}
	return (rc);
}

/*
 * _pjournal_logwritex - store a new entry in a journal transaction.
 * @pj: the journal.
 * @type: the application-specific log entry type.
 * @xid: transaction ID.
 * @data: the journal entry contents to store.
 * Returns: 0 on success, -1 on error.
 */
int
pjournal_logwrite(struct psc_journal_xidhndl *xh, int type, void *data,
		  size_t size)
{
	struct psc_journal_xidhndl	*t;
	int				 rc;
	struct psc_journal		*pj;
	uint32_t			 slot;
	uint32_t			 tail_slot;

	if (type & PJET_NODATA)
		psc_assert(!data);

	tail_slot = 0;
	pj = xh->pjx_pj;

	psc_assert(!(type & PJET_CORRUPT));
	psc_assert(!(type & PJET_XSTARTED));
	psc_assert(!(type & PJET_XEND));

	psc_assert(!(xh->pjx_flags & PJX_XCLOSED));

 retry:
	/*
	 * Make sure that the next slot to be written does not have a pending transaction.
	 * Since we add a new transaction at the tail of the pending transaction list, we
	 * only need to check the head of the list to find out the oldest pending transaction.
	 */
	PJ_LOCK(pj);
	slot = pj->pj_nextwrite;
	t = psclist_first_entry(&pj->pj_pndgxids, struct psc_journal_xidhndl, pjx_lentry);
	if (t) {
		if (t->pjx_tailslot == slot) {
			psc_warnx("pj(%p) blocking on slot(%d) "
				  "availability - owned by xid (%p)", 
				  pj, slot, t);
			psc_waitq_wait(&pj->pj_waitq, &pj->pj_lock);
			goto retry;
		}
		tail_slot = t->pjx_tailslot;
	}

	psc_info("pj(%p) tail@slot(%d) my@slot(%d) xh_flags(%o)",
		 pj, tail_slot, slot, xh->pjx_flags);

	if (!(xh->pjx_flags & PJET_XSTARTED)) {
		/* Multi-step operation, mark the slot id here
		 *  so that the tail of the journal can be found
		 *  and that overwriting pending xids may be
		 *  prevented.
		 * Note:  self-contained ops (PJET_XEND)
		 *        cannot end up here.
		 */
		psc_assert(!(xh->pjx_flags & PJET_XEND));

		xh->pjx_tailslot = slot;
		psclist_xadd_tail(&xh->pjx_lentry, &pj->pj_pndgxids);
		xh->pjx_flags |= PJET_XSTARTED;
	}
{'fraction_non_alphanumeric': 0.14554727344056492, 'fraction_numerical': 0.002746174970576697, 'mean_word_length': 4.0985, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 130, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4719883', 'n_tokens_mistral': 4896, 'n_tokens_neox': 4643, 'n_words': 968}