1/*-
2 * See the file LICENSE for redistribution information.
3 *
4 * Copyright (c) 1999-2009 Oracle.  All rights reserved.
5 *
6 * $Id$
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, "__qam_open: %s: unexpected file type or format",
94		    name);
95		ret = EINVAL;
96		goto err;
97	}
98
99	/* Setup information needed to open extents. */
100	t->page_ext = qmeta->page_ext;
101
102	if (t->page_ext != 0 && (ret = __qam_set_ext_data(dbp, name)) != 0)
103		goto err;
104
105	if (mode == 0)
106		mode = DB_MODE_660;
107	t->mode = mode;
108	t->re_pad = (int)qmeta->re_pad;
109	t->re_len = qmeta->re_len;
110	t->rec_page = qmeta->rec_page;
111
112	t->q_meta = base_pgno;
113	t->q_root = base_pgno + 1;
114
115err:	if (qmeta != NULL && (t_ret =
116	    __memp_fput(mpf, ip, qmeta, dbc->priority)) != 0 && ret == 0)
117		ret = t_ret;
118
119	/* Don't hold the meta page long term. */
120	if ((t_ret = __LPUT(dbc, metalock)) != 0 && ret == 0)
121		ret = t_ret;
122
123	if ((t_ret = __dbc_close(dbc)) != 0 && ret == 0)
124		ret = t_ret;
125
126	return (ret);
127}
128
129/*
130 * __qam_set_ext_data --
131 *	Setup DBP data for opening queue extents.
132 *
133 * PUBLIC: int __qam_set_ext_data __P((DB*, const char *));
134 */
135int
136__qam_set_ext_data(dbp, name)
137	DB *dbp;
138	const char *name;
139{
140	QUEUE *t;
141	int ret;
142
143	t = dbp->q_internal;
144	t->pginfo.db_pagesize = dbp->pgsize;
145	t->pginfo.flags =
146	    F_ISSET(dbp, (DB_AM_CHKSUM | DB_AM_ENCRYPT | DB_AM_SWAP));
147	t->pginfo.type = dbp->type;
148	t->pgcookie.data = &t->pginfo;
149	t->pgcookie.size = sizeof(DB_PGINFO);
150
151	if ((ret = __os_strdup(dbp->env, name,  &t->path)) != 0)
152		return (ret);
153	t->dir = t->path;
154	if ((t->name = __db_rpath(t->path)) == NULL) {
155		t->name = t->path;
156		t->dir = PATH_DOT;
157	} else
158		*t->name++ = '\0';
159
160	return (0);
161}
162
163/*
164 * __qam_metachk --
165 *
166 * PUBLIC: int __qam_metachk __P((DB *, const char *, QMETA *));
167 */
168int
169__qam_metachk(dbp, name, qmeta)
170	DB *dbp;
171	const char *name;
172	QMETA *qmeta;
173{
174	ENV *env;
175	u_int32_t vers;
176	int ret;
177
178	env = dbp->env;
179	ret = 0;
180
181	/*
182	 * At this point, all we know is that the magic number is for a Queue.
183	 * Check the version, the database may be out of date.
184	 */
185	vers = qmeta->dbmeta.version;
186	if (F_ISSET(dbp, DB_AM_SWAP))
187		M_32_SWAP(vers);
188	switch (vers) {
189	case 1:
190	case 2:
191		__db_errx(env,
192		    "%s: queue version %lu requires a version upgrade",
193		    name, (u_long)vers);
194		return (DB_OLD_VERSION);
195	case 3:
196	case 4:
197		break;
198	default:
199		__db_errx(env,
200		    "%s: unsupported qam version: %lu", name, (u_long)vers);
201		return (EINVAL);
202	}
203
204	/* Swap the page if we need to. */
205	if (F_ISSET(dbp, DB_AM_SWAP) &&
206	    (ret = __qam_mswap(env, (PAGE *)qmeta)) != 0)
207		return (ret);
208
209	/* Check the type. */
210	if (dbp->type != DB_QUEUE && dbp->type != DB_UNKNOWN)
211		return (EINVAL);
212	dbp->type = DB_QUEUE;
213	DB_ILLEGAL_METHOD(dbp, DB_OK_QUEUE);
214
215	/* Set the page size. */
216	dbp->pgsize = qmeta->dbmeta.pagesize;
217
218	/* Copy the file's ID. */
219	memcpy(dbp->fileid, qmeta->dbmeta.uid, DB_FILE_ID_LEN);
220
221	/* Set up AM-specific methods that do not require an open. */
222	dbp->db_am_rename = __qam_rename;
223	dbp->db_am_remove = __qam_remove;
224
225	return (ret);
226}
227
228/*
229 * __qam_init_meta --
230 *	Initialize the meta-data for a Queue database.
231 */
232static int
233__qam_init_meta(dbp, meta)
234	DB *dbp;
235	QMETA *meta;
236{
237	ENV *env;
238	QUEUE *t;
239
240	env = dbp->env;
241	t = dbp->q_internal;
242
243	memset(meta, 0, sizeof(QMETA));
244	LSN_NOT_LOGGED(meta->dbmeta.lsn);
245	meta->dbmeta.pgno = PGNO_BASE_MD;
246	meta->dbmeta.last_pgno = 0;
247	meta->dbmeta.magic = DB_QAMMAGIC;
248	meta->dbmeta.version = DB_QAMVERSION;
249	meta->dbmeta.pagesize = dbp->pgsize;
250	if (F_ISSET(dbp, DB_AM_CHKSUM))
251		FLD_SET(meta->dbmeta.metaflags, DBMETA_CHKSUM);
252	if (F_ISSET(dbp, DB_AM_ENCRYPT)) {
253		meta->dbmeta.encrypt_alg = env->crypto_handle->alg;
254		DB_ASSERT(env, meta->dbmeta.encrypt_alg != 0);
255		meta->crypto_magic = meta->dbmeta.magic;
256	}
257	meta->dbmeta.type = P_QAMMETA;
258	meta->re_pad = (u_int32_t)t->re_pad;
259	meta->re_len = t->re_len;
260	meta->rec_page = CALC_QAM_RECNO_PER_PAGE(dbp);
261	meta->cur_recno = 1;
262	meta->first_recno = 1;
263	meta->page_ext = t->page_ext;
264	t->rec_page = meta->rec_page;
265	memcpy(meta->dbmeta.uid, dbp->fileid, DB_FILE_ID_LEN);
266
267	/* Verify that we can fit at least one record per page. */
268	if (QAM_RECNO_PER_PAGE(dbp) < 1) {
269		__db_errx(env,
270		    "Record size of %lu too large for page size of %lu",
271		    (u_long)t->re_len, (u_long)dbp->pgsize);
272		return (EINVAL);
273	}
274
275	return (0);
276}
277
278/*
279 * __qam_new_file --
280 * Create the necessary pages to begin a new queue database file.
281 *
282 * PUBLIC: int __qam_new_file __P((DB *,
283 * PUBLIC:      DB_THREAD_INFO *, DB_TXN *, DB_FH *, const char *));
284 */
285int
286__qam_new_file(dbp, ip, txn, fhp, name)
287	DB *dbp;
288	DB_THREAD_INFO *ip;
289	DB_TXN *txn;
290	DB_FH *fhp;
291	const char *name;
292{
293	DBT pdbt;
294	DB_MPOOLFILE *mpf;
295	DB_PGINFO pginfo;
296	ENV *env;
297	QMETA *meta;
298	db_pgno_t pgno;
299	int ret, t_ret;
300
301	/*
302	 * Build meta-data page.
303	 *
304	 * This code appears more complex than it is because of the two cases
305	 * (named and unnamed).
306	 *
307	 * For each page being created, there are three parts: 1) a "get page"
308	 * chunk (which either uses malloc'd memory or calls __memp_fget), 2)
309	 * the initialization, and 3) the "put page" chunk which either does a
310	 * fop write or an __memp_fput.
311	 */
312	if (F_ISSET(dbp, DB_AM_INMEM)) {
313		mpf = dbp->mpf;
314		pgno = PGNO_BASE_MD;
315		if ((ret = __memp_fget(mpf, &pgno, ip, txn,
316		    DB_MPOOL_CREATE | DB_MPOOL_DIRTY, &meta)) != 0)
317			return (ret);
318
319		if ((ret = __qam_init_meta(dbp, meta)) != 0)
320			goto err1;
321
322		if ((ret = __db_log_page(dbp,
323		    txn, &meta->dbmeta.lsn, pgno, (PAGE *)meta)) != 0)
324			goto err1;
325err1:		if ((t_ret =
326		    __memp_fput(mpf, ip, meta, dbp->priority)) != 0 && ret == 0)
327			ret = t_ret;
328	} else {
329		env = dbp->env;
330		if ((ret = __os_calloc(env, 1, dbp->pgsize, &meta)) != 0)
331			return (ret);
332
333		if ((ret = __qam_init_meta(dbp, meta)) != 0)
334			goto err2;
335
336		pginfo.db_pagesize = dbp->pgsize;
337		pginfo.flags =
338		    F_ISSET(dbp, (DB_AM_CHKSUM | DB_AM_ENCRYPT | DB_AM_SWAP));
339		pginfo.type = DB_QUEUE;
340		DB_SET_DBT(pdbt, &pginfo, sizeof(pginfo));
341		if ((ret =
342		    __db_pgout(env->dbenv, PGNO_BASE_MD, meta, &pdbt)) != 0)
343			goto err2;
344		ret = __fop_write(env, txn, name, dbp->dirname,
345		    DB_APP_DATA, fhp, dbp->pgsize, 0, 0, meta, dbp->pgsize, 1,
346		    F_ISSET(dbp, DB_AM_NOT_DURABLE) ? DB_LOG_NOT_DURABLE : 0);
347
348err2:		__os_free(env, meta);
349	}
350
351	return (ret);
352}
353