<commit_before>// Copyright 2016 <NAME>
// MIT licensed (see LICENSE for details)

#include <assert.h>
#include <stdint.h>
#include <string.h>
#include <time.h>
#include <async/async.h>
#include <async/http/HTTP.h>
#include "util/hash.h"
#include "util/strext.h"
#include "util/url.h"
#include "db.h"
#include "page.h"
#include "errors.h"
#include "config.h"
#include "queue.h"

// fetch.c
int url_fetch(strarg_t const URL, strarg_t const client, struct response *const out);


static uint64_t current_id = 0;
static async_mutex_t id_lock[1];

static uint64_t work_time = 0;
static uint64_t work_id = 0;
static async_mutex_t work_lock[1];
static async_cond_t work_cond[1];

static async_mutex_t wait_lock[1];
static async_cond_t wait_cond[1];

// TODO: Define static async_x_t initializers
void queue_init(void) {
	async_mutex_init(id_lock, 0);
	async_mutex_init(work_lock, 0);
	async_cond_init(work_cond, 0);
	async_mutex_init(wait_lock, 0);
	async_cond_init(wait_cond, 0);
}


static int queue_peek(uint64_t *const outtime, uint64_t *const outid, char *const outURL, size_t const urlmax, char *const outclient, size_t const clientmax) {
	assert(outtime);
	assert(outid);
	assert(outURL);
	assert(urlmax > 0);
	assert(outclient);
	assert(clientmax > 0);

	DB_env *db = NULL;
	DB_txn *txn = NULL;
	DB_cursor *cursor = NULL;
	DB_range range[1];
	DB_val key[1];
	strarg_t URL, client;
	int rc = 0;

	rc = hx_db_open(&db);
	if(rc < 0) goto cleanup;
	rc = db_txn_begin(db, NULL, DB_RDONLY, &txn);
	if(rc < 0) goto cleanup;
	rc = db_txn_cursor(txn, &cursor);
	if(rc < 0) goto cleanup;
	HXTimeIDQueuedURLAndClientRange0(range);
	DB_VAL_STORAGE(key, DB_VARINT_MAX*3)
	db_bind_uint64(key, HXTimeIDQueuedURLAndClient);
	db_bind_uint64(key, work_time);
	db_bind_uint64(key, work_id+1);
	DB_VAL_STORAGE_VERIFY(key);
	rc = db_cursor_seekr(cursor, range, key, NULL, +1);
	if(rc < 0) goto cleanup;

	HXTimeIDQueuedURLAndClientKeyUnpack(key, txn, outtime, outid, &URL, &client);
	strlcpy(outURL, URL ? URL : "", urlmax);
	strlcpy(outclient, client ? client : "", clientmax);
	work_time = *outtime;
	work_id = *outid;
cleanup:
	cursor = NULL;
	db_txn_abort(txn); txn = NULL;
	hx_db_close(&db);
	return rc;
}
static int queue_remove(DB_txn *const txn, uint64_t const time, uint64_t const id, strarg_t const URL, strarg_t const client) {
	assert(time);
	assert(id);
	assert(URL);
	assert(client);
	char surt[URI_MAX];
	int rc = url_normalize_surt(URL, surt, sizeof(surt));
	if(rc < 0) goto cleanup;

	DB_val fwd_key[1];
	HXTimeIDQueuedURLAndClientKeyPack(fwd_key, txn, time, id, URL, client);
	rc = db_del(txn, fwd_key, 0);
	if(rc < 0) goto cleanup;

	DB_val rev_key[1];
	HXQueuedURLSurtAndTimeIDKeyPack(rev_key, txn, surt, time, id);
	rc = db_del(txn, rev_key, 0);
	if(rc < 0) goto cleanup;
cleanup:
	return rc;
}



int queue_add(uint64_t const time, strarg_t const URL, strarg_t const client) {
	assert(time);
	assert(URL);
	assert(client);
	char surt[URI_MAX];
	DB_env *db = NULL;
	DB_txn *txn = NULL;
	DB_cursor *cursor = NULL;
	int rc = 0;

	rc = url_normalize_surt(URL, surt, sizeof(surt));
	if(rc < 0) goto cleanup;

	async_mutex_lock(id_lock);
	uint64_t const id = ++current_id;
	async_mutex_unlock(id_lock);

	rc = hx_db_open(&db);
	if(rc < 0) goto cleanup;
	rc = db_txn_begin(db, NULL, DB_RDWR, &txn);
	if(rc < 0) goto cleanup;
	rc = db_txn_cursor(txn, &cursor);
	if(rc < 0) goto cleanup;

	DB_val chk_key[1];
	DB_range range_queued[1];
	HXQueuedURLSurtAndTimeIDRange1(range_queued, txn, surt);
	rc = db_cursor_firstr(cursor, range_queued, chk_key, NULL, -1);
	if(rc >= 0) goto cleanup; // If it's already queued, return success.
	if(DB_NOTFOUND != rc) goto cleanup;

	DB_range range_crawled[1];
	HXURLSurtAndTimeIDRange1(range_crawled, txn, surt);
	rc = db_cursor_firstr(cursor, range_crawled, chk_key, NULL, -1);
	if(rc >= 0) {
		strarg_t x;
		uint64_t ltime, lid;
		HXURLSurtAndTimeIDKeyUnpack(chk_key, txn, &x, &ltime, &lid);
		assert(0 == strcmp(x, surt));
		rc = ltime+CONFIG_CRAWL_DELAY_SECONDS < time ?
			DB_NOTFOUND : DB_KEYEXIST;
	}
	if(DB_NOTFOUND != rc) goto cleanup;

	DB_val fwd_key[1];
	HXTimeIDQueuedURLAndClientKeyPack(fwd_key, txn, time, id, URL, client);
	rc = db_put(txn, fwd_key, NULL, 0); // DB_NOOVERWRITE_FAST
	if(rc < 0) goto cleanup;

	DB_val rev_key[1];
	HXQueuedURLSurtAndTimeIDKeyPack(rev_key, txn, surt, time, id);
	rc = db_put(txn, rev_key, NULL, 0);
	if(rc < 0) goto cleanup;

	rc = db_txn_commit(txn); txn = NULL;
	if(rc < 0) goto cleanup;
	hx_db_close(&db);

	alogf("enqueued %s (%s)\n", URL, hx_strerror(rc));
	async_cond_broadcast(work_cond);
cleanup:
	cursor = NULL;
	db_txn_abort(txn); txn = NULL;
	hx_db_close(&db);
	return rc;
}
int queue_timedwait(uint64_t const time, strarg_t const URL, uint64_t const future) {
	DB_env *db = NULL;
	DB_txn *txn = NULL;
	int rc = 0;
	async_mutex_lock(wait_lock);
	for(;;) {
		// It was tempting to use double-checked locking
		// or something even fancier, but for now it doesn't
		// seem worth it.
		uint64_t ltime, lid;
		rc = hx_db_open(&db);
		if(rc < 0) break;
		rc = db_txn_begin(db, NULL, DB_RDONLY, &txn);
		if(rc < 0) break;
		rc = hx_get_latest(URL, txn, &ltime, &lid);
		db_txn_abort(txn); txn = NULL;
		hx_db_close(&db);
		if(rc >= 0 && ltime+CONFIG_CRAWL_DELAY_SECONDS >= time) {
			break;
		}
		if(DB_NOTFOUND != rc) break;
		rc = async_cond_timedwait(wait_cond, wait_lock, future);
		if(rc < 0) break;
	}
	async_mutex_unlock(wait_lock);
	db_txn_abort(txn); txn = NULL;
	hx_db_close(&db);
	return rc;
}

static void queue_work(void) {
	uint64_t then;
	uint64_t old_id;
	char URL[URI_MAX];
	char client[255+1];
<commit_msg>Tweak queue_timedwait(). Existing entry too old -> not found, keep waiting.<commit_after>// Copyright 2016 <NAME>
// MIT licensed (see LICENSE for details)

#include <assert.h>
#include <stdint.h>
#include <string.h>
#include <time.h>
#include <async/async.h>
#include <async/http/HTTP.h>
#include "util/hash.h"
#include "util/strext.h"
#include "util/url.h"
#include "db.h"
#include "page.h"
#include "errors.h"
#include "config.h"
#include "queue.h"

// fetch.c
int url_fetch(strarg_t const URL, strarg_t const client, struct response *const out);


static uint64_t current_id = 0;
static async_mutex_t id_lock[1];

static uint64_t work_time = 0;
static uint64_t work_id = 0;
static async_mutex_t work_lock[1];
static async_cond_t work_cond[1];

static async_mutex_t wait_lock[1];
static async_cond_t wait_cond[1];

// TODO: Define static async_x_t initializers
void queue_init(void) {
	async_mutex_init(id_lock, 0);
	async_mutex_init(work_lock, 0);
	async_cond_init(work_cond, 0);
	async_mutex_init(wait_lock, 0);
	async_cond_init(wait_cond, 0);
}


static int queue_peek(uint64_t *const outtime, uint64_t *const outid, char *const outURL, size_t const urlmax, char *const outclient, size_t const clientmax) {
	assert(outtime);
	assert(outid);
	assert(outURL);
	assert(urlmax > 0);
	assert(outclient);
	assert(clientmax > 0);

	DB_env *db = NULL;
	DB_txn *txn = NULL;
	DB_cursor *cursor = NULL;
	DB_range range[1];
	DB_val key[1];
	strarg_t URL, client;
	int rc = 0;

	rc = hx_db_open(&db);
	if(rc < 0) goto cleanup;
	rc = db_txn_begin(db, NULL, DB_RDONLY, &txn);
	if(rc < 0) goto cleanup;
	rc = db_txn_cursor(txn, &cursor);
	if(rc < 0) goto cleanup;
	HXTimeIDQueuedURLAndClientRange0(range);
	DB_VAL_STORAGE(key, DB_VARINT_MAX*3)
	db_bind_uint64(key, HXTimeIDQueuedURLAndClient);
	db_bind_uint64(key, work_time);
	db_bind_uint64(key, work_id+1);
	DB_VAL_STORAGE_VERIFY(key);
	rc = db_cursor_seekr(cursor, range, key, NULL, +1);
	if(rc < 0) goto cleanup;

	HXTimeIDQueuedURLAndClientKeyUnpack(key, txn, outtime, outid, &URL, &client);
	strlcpy(outURL, URL ? URL : "", urlmax);
	strlcpy(outclient, client ? client : "", clientmax);
	work_time = *outtime;
	work_id = *outid;
cleanup:
	cursor = NULL;
	db_txn_abort(txn); txn = NULL;
	hx_db_close(&db);
	return rc;
}
static int queue_remove(DB_txn *const txn, uint64_t const time, uint64_t const id, strarg_t const URL, strarg_t const client) {
	assert(time);
	assert(id);
	assert(URL);
	assert(client);
	char surt[URI_MAX];
	int rc = url_normalize_surt(URL, surt, sizeof(surt));
	if(rc < 0) goto cleanup;

	DB_val fwd_key[1];
	HXTimeIDQueuedURLAndClientKeyPack(fwd_key, txn, time, id, URL, client);
	rc = db_del(txn, fwd_key, 0);
	if(rc < 0) goto cleanup;

	DB_val rev_key[1];
	HXQueuedURLSurtAndTimeIDKeyPack(rev_key, txn, surt, time, id);
	rc = db_del(txn, rev_key, 0);
	if(rc < 0) goto cleanup;
cleanup:
	return rc;
}



int queue_add(uint64_t const time, strarg_t const URL, strarg_t const client) {
	assert(time);
	assert(URL);
	assert(client);
	char surt[URI_MAX];
	DB_env *db = NULL;
	DB_txn *txn = NULL;
	DB_cursor *cursor = NULL;
	int rc = 0;

	rc = url_normalize_surt(URL, surt, sizeof(surt));
	if(rc < 0) goto cleanup;

	async_mutex_lock(id_lock);
	uint64_t const id = ++current_id;
	async_mutex_unlock(id_lock);

	rc = hx_db_open(&db);
	if(rc < 0) goto cleanup;
	rc = db_txn_begin(db, NULL, DB_RDWR, &txn);
	if(rc < 0) goto cleanup;
	rc = db_txn_cursor(txn, &cursor);
	if(rc < 0) goto cleanup;

	DB_val chk_key[1];
	DB_range range_queued[1];
	HXQueuedURLSurtAndTimeIDRange1(range_queued, txn, surt);
	rc = db_cursor_firstr(cursor, range_queued, chk_key, NULL, -1);
	if(rc >= 0) goto cleanup; // If it's already queued, return success.
	if(DB_NOTFOUND != rc) goto cleanup;

	DB_range range_crawled[1];
	HXURLSurtAndTimeIDRange1(range_crawled, txn, surt);
	rc = db_cursor_firstr(cursor, range_crawled, chk_key, NULL, -1);
	if(rc >= 0) {
		strarg_t x;
		uint64_t ltime, lid;
		HXURLSurtAndTimeIDKeyUnpack(chk_key, txn, &x, &ltime, &lid);
		assert(0 == strcmp(x, surt));
		rc = ltime+CONFIG_CRAWL_DELAY_SECONDS < time ?
			DB_NOTFOUND : DB_KEYEXIST;
	}
	if(DB_NOTFOUND != rc) goto cleanup;

	DB_val fwd_key[1];
	HXTimeIDQueuedURLAndClientKeyPack(fwd_key, txn, time, id, URL, client);
	rc = db_put(txn, fwd_key, NULL, 0); // DB_NOOVERWRITE_FAST
	if(rc < 0) goto cleanup;

	DB_val rev_key[1];
	HXQueuedURLSurtAndTimeIDKeyPack(rev_key, txn, surt, time, id);
	rc = db_put(txn, rev_key, NULL, 0);
	if(rc < 0) goto cleanup;

	rc = db_txn_commit(txn); txn = NULL;
	if(rc < 0) goto cleanup;
	hx_db_close(&db);

	alogf("enqueued %s (%s)\n", URL, hx_strerror(rc));
	async_cond_broadcast(work_cond);
cleanup:
	cursor = NULL;
	db_txn_abort(txn); txn = NULL;
	hx_db_close(&db);
	return rc;
}
int queue_timedwait(uint64_t const time, strarg_t const URL, uint64_t const future) {
	DB_env *db = NULL;
	DB_txn *txn = NULL;
	int rc = 0;
	async_mutex_lock(wait_lock);
	for(;;) {
		// It was tempting to use double-checked locking
		// or something even fancier, but for now it doesn't
		// seem worth it.
		uint64_t ltime, lid;
		rc = hx_db_open(&db);
		if(rc < 0) break;
		rc = db_txn_begin(db, NULL, DB_RDONLY, &txn);
		if(rc < 0) break;
		rc = hx_get_latest(URL, txn, &ltime, &lid);
		db_txn_abort(txn); txn = NULL;
		hx_db_close(&db);
		if(rc >= 0) {
			if(ltime+CONFIG_CRAWL_DELAY_SECONDS >= time) break;
			rc = DB_NOTFOUND;
		}
		if(DB_NOTFOUND != rc) break;
		rc = async_cond_timedwait(wait_cond, wait_lock, future);
		if(rc < 0) break;
	}
	async_mutex_unlock(wait_lock);
	db_txn_abort(txn); txn = NULL;
	hx_db_close(&db);
	return rc;
}

static void queue_work(void) {
	uint64_t then;
	uint64_t old_id;
	char URL[URI_MAX];
	char client[255+1];

{'fraction_non_alphanumeric': 0.13568773234200743, 'fraction_numerical': 0.018410338112940343, 'mean_word_length': 5.191232876712329, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 30, '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': '6362021', 'n_tokens_mistral': 5085, 'n_tokens_neox': 4868, 'n_words': 1125}