1/*-
2 * See the file LICENSE for redistribution information.
3 *
4 * Copyright (c) 1999,2008 Oracle.  All rights reserved.
5 *
6 * $Id: qam_open.c,v 12.24 2008/01/30 12:18:23 mjc Exp $
7 */
8
9#include "db_config.h"
10
11#include "db_int.h"
12#include "dbinc/crypto.h"
13#include "dbinc/db_page.h"
14#include "dbinc/db_swap.h"
15#include "dbinc/db_am.h"
16#include "dbinc/lock.h"
17#include "dbinc/mp.h"
18#include "dbinc/qam.h"
19#include "dbinc/fop.h"
20
21static int __qam_init_meta __P((DB *, QMETA *));
22
23/*
24 * __qam_open
25 *
26 * PUBLIC: int __qam_open __P((DB *, DB_THREAD_INFO *,
27 * PUBLIC:     DB_TXN *, const char *, db_pgno_t, int, u_int32_t));
28 */
29int
30__qam_open(dbp, ip, txn, name, base_pgno, mode, flags)
31	DB *dbp;
32	DB_THREAD_INFO *ip;
33	DB_TXN *txn;
34	const char *name;
35	db_pgno_t base_pgno;
36	int mode;
37	u_int32_t flags;
38{
39	DBC *dbc;
40	DB_LOCK metalock;
41	DB_MPOOLFILE *mpf;
42	ENV *env;
43	QMETA *qmeta;
44	QUEUE *t;
45	int ret, t_ret;
46
47	env = dbp->env;
48	mpf = dbp->mpf;
49	t = dbp->q_internal;
50	ret = 0;
51	qmeta = NULL;
52
53	if (name == NULL && t->page_ext != 0) {
54		__db_errx(env,
55	"Extent size may not be specified for in-memory queue database");
56		return (EINVAL);
57	}
58
59	if (MULTIVERSION(dbp)) {
60		__db_errx(env,
61		    "Multiversion queue databases are not supported");
62		return (EINVAL);
63	}
64
65	/* Initialize the remaining fields/methods of the DB. */
66	dbp->db_am_remove = __qam_remove;
67	dbp->db_am_rename = __qam_rename;
68
69	/*
70	 * Get a cursor.  If DB_CREATE is specified, we may be creating
71	 * pages, and to do that safely in CDB we need a write cursor.
72	 * In STD_LOCKING mode, we'll synchronize using the meta page
73	 * lock instead.
74	 */
75	if ((ret = __db_cursor(dbp, ip, txn, &dbc,
76	    LF_ISSET(DB_CREATE) && CDB_LOCKING(env) ?
77	    DB_WRITECURSOR : 0)) != 0)
78		return (ret);
79
80	/*
81	 * Get the meta data page.  It must exist, because creates of
82	 * files/databases come in through the __qam_new_file interface
83	 * and queue doesn't support subdatabases.
84	 */
85	if ((ret =
86	    __db_lget(dbc, 0, base_pgno, DB_LOCK_READ, 0, &metalock)) != 0)
87		goto err;
88	if ((ret = __memp_fget(mpf, &base_pgno, ip, txn, 0, &qmeta)) != 0)
89		goto err;
90
91	/* If the magic number is incorrect, that's a fatal error. */
92	if (qmeta->dbmeta.magic != DB_QAMMAGIC) {
93		__db_errx(env, "%s: unexpected file type or format", name);
94		ret = EINVAL;
95		goto err;
96	}
97
98	/* Setup information needed to open extents. */
99	t->page_ext = qmeta->page_ext;
100
101	if (t->page_ext != 0 && (ret = __qam_set_ext_data(dbp, name)) != 0)
102		goto err;
103
104	if (mode == 0)
105		mode = DB_MODE_660;
106	t->mode = mode;
107	t->re_pad = (int)qmeta->re_pad;
108	t->re_len = qmeta->re_len;
109	t->rec_page = qmeta->rec_page;
110
111	t->q_meta = base_pgno;
112	t->q_root = base_pgno + 1;
113
114err:	if (qmeta != NULL && (t_ret =
115	    __memp_fput(mpf, ip, qmeta, dbc->priority)) != 0 && ret == 0)
116		ret = t_ret;
117
118	/* Don't hold the meta page long term. */
119	if ((t_ret = __LPUT(dbc, metalock)) != 0 && ret == 0)
120		ret = t_ret;
121
122	if ((t_ret = __dbc_close(dbc)) != 0 && ret == 0)
123		ret = t_ret;
124
125	return (ret);
126}
127
128/*
129 * __qam_set_ext_data --
130 *	Setup DBP data for opening queue extents.
131 *
132 * PUBLIC: int __qam_set_ext_data __P((DB*, const char *));
133 */
134int
135__qam_set_ext_data(dbp, name)
136	DB *dbp;
137	const char *name;
138{
139	QUEUE *t;
140	int ret;
141
142	t = dbp->q_internal;
143	t->pginfo.db_pagesize = dbp->pgsize;
144	t->pginfo.flags =
145	    F_ISSET(dbp, (DB_AM_CHKSUM | DB_AM_ENCRYPT | DB_AM_SWAP));
146	t->pginfo.type = dbp->type;
147	t->pgcookie.data = &t->pginfo;
148	t->pgcookie.size = sizeof(DB_PGINFO);
149
150	if ((ret = __os_strdup(dbp->env, name,  &t->path)) != 0)
151		return (ret);
152	t->dir = t->path;
153	if ((t->name = __db_rpath(t->path)) == NULL) {
154		t->name = t->path;
155		t->dir = PATH_DOT;
156	} else
157		*t->name++ = '\0';
158
159	return (0);
160}
161
162/*
163 * __qam_metachk --
164 *
165 * PUBLIC: int __qam_metachk __P((DB *, const char *, QMETA *));
166 */
167int
168__qam_metachk(dbp, name, qmeta)
169	DB *dbp;
170	const char *name;
171	QMETA *qmeta;
172{
173	ENV *env;
174	u_int32_t vers;
175	int ret;
176
177	env = dbp->env;
178	ret = 0;
179
180	/*
181	 * At this point, all we know is that the magic number is for a Queue.
182	 * Check the version, the database may be out of date.
183	 */
184	vers = qmeta->dbmeta.version;
185	if (F_ISSET(dbp, DB_AM_SWAP))
186		M_32_SWAP(vers);
187	switch (vers) {
188	case 1:
189	case 2:
190		__db_errx(env,
191		    "%s: queue version %lu requires a version upgrade",
192		    name, (u_long)vers);
193		return (DB_OLD_VERSION);
194	case 3:
195	case 4:
196		break;
197	default:
198		__db_errx(env,
199		    "%s: unsupported qam version: %lu", name, (u_long)vers);
200		return (EINVAL);
201	}
202
203	/* Swap the page if we need to. */
204	if (F_ISSET(dbp, DB_AM_SWAP) &&
205	    (ret = __qam_mswap(env, (PAGE *)qmeta)) != 0)
206		return (ret);
207
208	/* Check the type. */
209	if (dbp->type != DB_QUEUE && dbp->type != DB_UNKNOWN)
210		return (EINVAL);
211	dbp->type = DB_QUEUE;
212	DB_ILLEGAL_METHOD(dbp, DB_OK_QUEUE);
213
214	/* Set the page size. */
215	dbp->pgsize = qmeta->dbmeta.pagesize;
216
217	/* Copy the file's ID. */
218	memcpy(dbp->fileid, qmeta->dbmeta.uid, DB_FILE_ID_LEN);
219
220	/* Set up AM-specific methods that do not require an open. */
221	dbp->db_am_rename = __qam_rename;
222	dbp->db_am_remove = __qam_remove;
223
224	return (ret);
225}
226
227/*
228 * __qam_init_meta --
229 *	Initialize the meta-data for a Queue database.
230 */
231static int
232__qam_init_meta(dbp, meta)
233	DB *dbp;
234	QMETA *meta;
235{
236	ENV *env;
237	QUEUE *t;
238
239	env = dbp->env;
240	t = dbp->q_internal;
241
242	memset(meta, 0, sizeof(QMETA));
243	LSN_NOT_LOGGED(meta->dbmeta.lsn);
244	meta->dbmeta.pgno = PGNO_BASE_MD;
245	meta->dbmeta.last_pgno = 0;
246	meta->dbmeta.magic = DB_QAMMAGIC;
247	meta->dbmeta.version = DB_QAMVERSION;
248	meta->dbmeta.pagesize = dbp->pgsize;
249	if (F_ISSET(dbp, DB_AM_CHKSUM))
250		FLD_SET(meta->dbmeta.metaflags, DBMETA_CHKSUM);
251	if (F_ISSET(dbp, DB_AM_ENCRYPT)) {
252		meta->dbmeta.encrypt_alg = env->crypto_handle->alg;
253		DB_ASSERT(env, meta->dbmeta.encrypt_alg != 0);
254		meta->crypto_magic = meta->dbmeta.magic;
255	}
256	meta->dbmeta.type = P_QAMMETA;
257	meta->re_pad = (u_int32_t)t->re_pad;
258	meta->re_len = t->re_len;
259	meta->rec_page = CALC_QAM_RECNO_PER_PAGE(dbp);
260	meta->cur_recno = 1;
261	meta->first_recno = 1;
262	meta->page_ext = t->page_ext;
263	t->rec_page = meta->rec_page;
264	memcpy(meta->dbmeta.uid, dbp->fileid, DB_FILE_ID_LEN);
265
266	/* Verify that we can fit at least one record per page. */
267	if (QAM_RECNO_PER_PAGE(dbp) < 1) {
268		__db_errx(env,
269		    "Record size of %lu too large for page size of %lu",
270		    (u_long)t->re_len, (u_long)dbp->pgsize);
271		return (EINVAL);
272	}
273
274	return (0);
275}
276
277/*
278 * __qam_new_file --
279 * Create the necessary pages to begin a new queue database file.
280 *
281 * PUBLIC: int __qam_new_file __P((DB *,
282 * PUBLIC:      DB_THREAD_INFO *, DB_TXN *, DB_FH *, const char *));
283 */
284int
285__qam_new_file(dbp, ip, txn, fhp, name)
286	DB *dbp;
287	DB_THREAD_INFO *ip;
288	DB_TXN *txn;
289	DB_FH *fhp;
290	const char *name;
291{
292	DBT pdbt;
293	DB_MPOOLFILE *mpf;
294	DB_PGINFO pginfo;
295	ENV *env;
296	QMETA *meta;
297	db_pgno_t pgno;
298	int ret, t_ret;
299
300	/*
301	 * Build meta-data page.
302	 *
303	 * This code appears more complex than it is because of the two cases
304	 * (named and unnamed).
305	 *
306	 * For each page being created, there are three parts: 1) a "get page"
307	 * chunk (which either uses malloc'd memory or calls __memp_fget), 2)
308	 * the initialization, and 3) the "put page" chunk which either does a
309	 * fop write or an __memp_fput.
310	 */
311	if (F_ISSET(dbp, DB_AM_INMEM)) {
312		mpf = dbp->mpf;
313		pgno = PGNO_BASE_MD;
314		if ((ret = __memp_fget(mpf, &pgno, ip, txn,
315		    DB_MPOOL_CREATE | DB_MPOOL_DIRTY, &meta)) != 0)
316			return (ret);
317
318		if ((ret = __qam_init_meta(dbp, meta)) != 0)
319			goto err1;
320
321		if ((ret = __db_log_page(dbp,
322		    txn, &meta->dbmeta.lsn, pgno, (PAGE *)meta)) != 0)
323			goto err1;
324err1:		if ((t_ret =
325		    __memp_fput(mpf, ip, meta, dbp->priority)) != 0 && ret == 0)
326			ret = t_ret;
327	} else {
328		env = dbp->env;
329		if ((ret = __os_calloc(env, 1, dbp->pgsize, &meta)) != 0)
330			return (ret);
331
332		if ((ret = __qam_init_meta(dbp, meta)) != 0)
333			goto err2;
334
335		pginfo.db_pagesize = dbp->pgsize;
336		pginfo.flags =
337		    F_ISSET(dbp, (DB_AM_CHKSUM | DB_AM_ENCRYPT | DB_AM_SWAP));
338		pginfo.type = DB_QUEUE;
339		DB_SET_DBT(pdbt, &pginfo, sizeof(pginfo));
340		if ((ret =
341		    __db_pgout(env->dbenv, PGNO_BASE_MD, meta, &pdbt)) != 0)
342			goto err2;
343		ret = __fop_write(env, txn, name,
344		    DB_APP_DATA, fhp, dbp->pgsize, 0, 0, meta, dbp->pgsize, 1,
345		    F_ISSET(dbp, DB_AM_NOT_DURABLE) ? DB_LOG_NOT_DURABLE : 0);
346
347err2:		__os_free(env, meta);
348	}
349
350	return (ret);
351}
352