<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);

	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(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);

	psc_assert(!pje->pje_genmarker);

	pje->pje_genmarker |= pj->pj_genid;
	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);

	//XXX	rc = pwrite(pj->pj_fd, pje, pj->pj_hdr->pjh_entsz,
	//XXX       (off_t)(pj->pj_hdr->pjh_start_off + (slot * pj->pj_hdr->pjh_entsz)));

#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);
}<commit_msg>uncomment pwrite() in pjournal_logwrite()<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);

	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(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, 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);

	psc_assert(!pje->pje_genmarker);

	pje->pje_genmarker |= pj->pj_genid;
	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);

	/* 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)));

#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);
}
{'fraction_non_alphanumeric': 0.1577713667508114, 'fraction_numerical': 0.001983411467724486, 'mean_word_length': 4.919957310565635, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 89, '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': '4719882', 'n_tokens_mistral': 2754, 'n_tokens_neox': 2682, 'n_words': 470}