1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
#include <windows.h>
#include <clfsw32.h>
#include <clfsmgmtw32.h>
#include <sstream>
#include <string>
#include <vector>
#include <stdlib.h>
#include <qpid/sys/windows/check.h>
#include "Log.h"
namespace qpid {
namespace store {
namespace ms_clfs {
Log::~Log()
{
if (marshal != 0)
::DeleteLogMarshallingArea(marshal);
::CloseHandle(handle);
}
void
Log::open(const std::string& path, const TuningParameters& params)
{
this->containerSize = static_cast<ULONGLONG>(params.containerSize);
logPath = path;
std::string logSpec = "log:" + path;
size_t specLength = logSpec.length();
std::auto_ptr<wchar_t> wLogSpec(new wchar_t[specLength + 1]);
size_t converted;
mbstowcs_s(&converted,
wLogSpec.get(), specLength+1,
logSpec.c_str(), specLength);
handle = ::CreateLogFile(wLogSpec.get(),
GENERIC_WRITE | GENERIC_READ,
0,
0,
OPEN_ALWAYS,
0);
QPID_WINDOWS_CHECK_NOT(handle, INVALID_HANDLE_VALUE);
CLFS_INFORMATION info;
ULONG infoSize = sizeof(info);
BOOL ok = ::GetLogFileInformation(handle, &info, &infoSize);
QPID_WINDOWS_CHECK_NOT(ok, 0);
ok = ::RegisterManageableLogClient(handle, 0);
QPID_WINDOWS_CHECK_NOT(ok, 0);
// Set up policies for how many containers to initially create and how
// large each container should be. Also, auto-grow the log when container
// space runs out.
CLFS_MGMT_POLICY logPolicy;
logPolicy.Version = CLFS_MGMT_POLICY_VERSION;
logPolicy.LengthInBytes = sizeof(logPolicy);
logPolicy.PolicyFlags = 0;
// If this is the first time this log is opened, give an opportunity to
// initialize its content.
bool needInitialize(false);
if (info.TotalContainers == 0) {
// New log; set the configured container size and create the
// initial set of containers.
logPolicy.PolicyType = ClfsMgmtPolicyNewContainerSize;
logPolicy.PolicyParameters.NewContainerSize.SizeInBytes = containerSize;
ok = ::InstallLogPolicy(handle, &logPolicy);
QPID_WINDOWS_CHECK_NOT(ok, 0);
ULONGLONG desired(params.containers), actual(0);
ok = ::SetLogFileSizeWithPolicy(handle, &desired, &actual);
QPID_WINDOWS_CHECK_NOT(ok, 0);
needInitialize = true;
}
// Ensure that the log is extended as needed and will shrink when 50%
// becomes unused.
logPolicy.PolicyType = ClfsMgmtPolicyAutoGrow;
logPolicy.PolicyParameters.AutoGrow.Enabled = 1;
ok = ::InstallLogPolicy(handle, &logPolicy);
QPID_WINDOWS_CHECK_NOT(ok, 0);
logPolicy.PolicyType = ClfsMgmtPolicyAutoShrink;
logPolicy.PolicyParameters.AutoShrink.Percentage = params.shrinkPct;
ok = ::InstallLogPolicy(handle, &logPolicy);
QPID_WINDOWS_CHECK_NOT(ok, 0);
// Need a marshaling area
ok = ::CreateLogMarshallingArea(handle,
NULL, NULL, NULL, // Alloc, free, context
marshallingBufferSize(),
params.maxWriteBuffers,
1, // Max read buffers
&marshal);
QPID_WINDOWS_CHECK_NOT(ok, 0);
if (needInitialize)
initialize();
}
uint32_t
Log::marshallingBufferSize()
{
// Default implementation returns the minimum marshalling buffer size;
// derived ones should come up with a more fitting value.
//
// Find the directory name part of the log specification, including the
// trailing '\'.
size_t dirMarker = logPath.rfind('\\');
if (dirMarker == std::string::npos)
dirMarker = logPath.rfind('/');
DWORD bytesPerSector;
DWORD dontCare;
::GetDiskFreeSpace(logPath.substr(0, dirMarker).c_str(),
&dontCare,
&bytesPerSector,
&dontCare,
&dontCare);
return bytesPerSector;
}
CLFS_LSN
Log::write(void* entry, uint32_t length, CLFS_LSN* prev)
{
CLFS_WRITE_ENTRY desc;
desc.Buffer = entry;
desc.ByteLength = length;
CLFS_LSN lsn;
BOOL ok = ::ReserveAndAppendLog(marshal,
&desc, 1, // Buffer descriptor
0, prev, // Undo-Next, Prev
0, 0, // Reservation
CLFS_FLAG_FORCE_FLUSH,
&lsn,
0);
QPID_WINDOWS_CHECK_NOT(ok, 0);
return lsn;
}
// Get the current base LSN of the log.
CLFS_LSN
Log::getBase()
{
CLFS_INFORMATION info;
ULONG infoSize = sizeof(info);
BOOL ok = ::GetLogFileInformation(handle, &info, &infoSize);
QPID_WINDOWS_CHECK_NOT(ok, 0);
return info.BaseLsn;
}
void
Log::moveTail(const CLFS_LSN& oldest)
{
BOOL ok = ::AdvanceLogBase(marshal,
const_cast<PCLFS_LSN>(&oldest),
0, NULL);
// If multiple threads are manipulating things they may get out of
// order when moving the tail; if someone already moved it further
// than this, it's ok - ignore it.
if (ok || ::GetLastError() == ERROR_LOG_START_OF_LOG)
return;
QPID_WINDOWS_CHECK_NOT(ok, 0);
}
}}} // namespace qpid::store::ms_clfs
|
