/* 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.
*/
#define C_LUCY_INSTREAM
#define C_LUCY_FILEWINDOW
#include "Lucy/Util/ToolSet.h"
#include "Lucy/Store/InStream.h"
#include "Lucy/Store/FileHandle.h"
#include "Lucy/Store/FSFileHandle.h"
#include "Lucy/Store/FileWindow.h"
#include "Lucy/Store/RAMFile.h"
#include "Lucy/Store/RAMFileHandle.h"
// Inlined version of InStream_Tell.
static INLINE int64_t
SI_tell(InStream *self);
// Inlined version of InStream_Read_Bytes.
static INLINE void
SI_read_bytes(InStream *self, char* buf, size_t len);
// Inlined version of InStream_Read_U8.
static INLINE uint8_t
SI_read_u8(InStream *self);
// Ensure that the buffer contains exactly the specified amount of data.
static void
S_fill(InStream *self, int64_t amount);
// Refill the buffer, with either IO_STREAM_BUF_SIZE bytes or all remaining
// file content -- whichever is smaller. Throw an error if we're at EOF and
// can't load at least one byte.
static int64_t
S_refill(InStream *self);
InStream*
InStream_open(Obj *file) {
InStream *self = (InStream*)VTable_Make_Obj(INSTREAM);
return InStream_do_open(self, file);
}
InStream*
InStream_do_open(InStream *self, Obj *file) {
// Init.
self->buf = NULL;
self->limit = NULL;
self->offset = 0;
self->window = FileWindow_new();
// Obtain a FileHandle.
if (Obj_Is_A(file, FILEHANDLE)) {
self->file_handle = (FileHandle*)INCREF(file);
}
else if (Obj_Is_A(file, RAMFILE)) {
self->file_handle
= (FileHandle*)RAMFH_open(NULL, FH_READ_ONLY, (RAMFile*)file);
}
else if (Obj_Is_A(file, CHARBUF)) {
self->file_handle
= (FileHandle*)FSFH_open((CharBuf*)file, FH_READ_ONLY);
}
else {
Err_set_error(Err_new(CB_newf("Invalid type for param 'file': '%o'",
Obj_Get_Class_Name(file))));
DECREF(self);
return NULL;
}
if (!self->file_handle) {
ERR_ADD_FRAME(Err_get_error());
DECREF(self);
return NULL;
}
// Get length and filename from the FileHandle.
self->filename = CB_Clone(FH_Get_Path(self->file_handle));
self->len = FH_Length(self->file_handle);
if (self->len == -1) {
ERR_ADD_FRAME(Err_get_error());
DECREF(self);
return NULL;
}
return self;
}
void
InStream_close(InStream *self) {
if (self->file_handle) {
FH_Release_Window(self->file_handle, self->window);
// Note that we don't close the FileHandle, because it's probably
// shared.
DECREF(self->file_handle);
self->file_handle = NULL;
}
}
void
InStream_destroy(InStream *self) {
if (self->file_handle) {
InStream_Close(self);
}
DECREF(self->filename);
DECREF(self->window);
SUPER_DESTROY(self, INSTREAM);
}
InStream*
InStream_reopen(InStream *self, const CharBuf *filename, int64_t offset,
int64_t len) {
if (!self->file_handle) {
THROW(ERR, "Can't Reopen() closed InStream %o", self->filename);
}
if (offset + len > FH_Length(self->file_handle)) {
THROW(ERR, "Offset + length too large (%i64 + %i64 > %i64)",
offset, len, FH_Length(self->file_handle));
}
InStream *twin = (InStream*)VTable_Make_Obj(self->vtable);
InStream_do_open(twin, (Obj*)self->file_handle);
if (filename != NULL) { CB_Mimic(twin->filename, (Obj*)filename); }
twin->offset = offset;
twin->len = len;
InStream_Seek(twin, 0);
return twin;
}
InStream*
InStream_clone(InStream *self) {
InStream *twin = (InStream*)VTable_Make_Obj(self->vtable);
InStream_do_open(twin, (Obj*)self->file_handle);
InStream_Seek(twin, SI_tell(self));
return twin;
}
CharBuf*
InStream_get_filename(InStream *self) {
return self->filename;
}
static int64_t
S_refill(InStream *self) {
// Determine the amount to request.
const int64_t sub_file_pos = SI_tell(self);
const int64_t remaining = self->len - sub_file_pos;
const int64_t amount = remaining < IO_STREAM_BUF_SIZE
? remaining
: IO_STREAM_BUF_SIZE;
if (!remaining) {
THROW(ERR, "Read past EOF of '%o' (offset: %i64 len: %i64)",
self->filename, self->offset, self->len);
}
// Make the request.
S_fill(self, amount);
return amount;
}
void
InStream_refill(InStream *self) {
S_refill(self);
}
static void
S_fill(InStream *self, int64_t amount) {
FileWindow *const window = self->window;
const int64_t virtual_file_pos = SI_tell(self);
const int64_t real_file_pos = virtual_file_pos + self->offset;
const int64_t remaining = self->len - virtual_file_pos;
// Throw an error if the requested amount would take us beyond EOF.
if (amount > remaining) {
THROW(ERR, "Read past EOF of %o (pos: %u64 len: %u64 request: %u64)",
self->filename, virtual_file_pos, self->len, amount);
}
// Make the request.
if (FH_Window(self->file_handle, window, real_file_pos, amount)) {
char *const window_limit = window->buf + window->len;
self->buf = window->buf
- window->offset // theoretical start of real file
+ self->offset // top of virtual file
+ virtual_file_pos; // position within virtual file
self->limit = window_limit - self->buf > remaining
? self->buf + remaining
: window_limit;
}
else {
Err *error = Err_get_error();
CB_catf(Err_Get_Mess(error), " (%o)", self->filename);
RETHROW(INCREF(error));
}
}
void
InStream_fill(InStream *self, int64_t amount) {
S_fill(self, amount);
}
void
InStream_seek(InStream *self, int64_t target) {
FileWindow *const window = self->window;
int64_t virtual_window_top = window->offset - self->offset;
int64_t virtual_window_end = virtual_window_top + window->len;
if (target < 0) {
THROW(ERR, "Can't Seek '%o' to negative target %i64", self->filename,
target);
}
// Seek within window if possible.
else if (target >= virtual_window_top
&& target <= virtual_window_end
) {
self->buf = window->buf - window->offset + self->offset + target;
}
else if (target > self->len) {
THROW(ERR, "Can't Seek '%o' past EOF (%i64 > %i64)", self->filename,
target, self->len);
}
else {
// Target is outside window. Set all buffer and limit variables to
// NULL to trigger refill on the next read. Store the file position
// in the FileWindow's offset.
FH_Release_Window(self->file_handle, window);
self->buf = NULL;
self->limit = NULL;
FileWindow_Set_Offset(window, self->offset + target);
}
}
static INLINE int64_t
SI_tell(InStream *self) {
FileWindow *const window = self->window;
int64_t pos_in_buf = PTR_TO_I64(self->buf) - PTR_TO_I64(window->buf);
return pos_in_buf + window->offset - self->offset;
}
int64_t
InStream_tell(InStream *self) {
return SI_tell(self);
}
int64_t
InStream_length(InStream *self) {
return self->len;
}
char*
InStream_buf(InStream *self, size_t request) {
const int64_t bytes_in_buf = PTR_TO_I64(self->limit) - PTR_TO_I64(self->buf);
/* It's common for client code to overestimate how much is needed, because
* the request has to figure in worst-case for compressed data. However,
* if we can still serve them everything they request (e.g. they ask for 5
* bytes, they really need 1 byte, and there's 1k in the buffer), we can
* skip the following refill block. */
if ((int64_t)request > bytes_in_buf) {
const int64_t remaining_in_file = self->len - SI_tell(self);
int64_t amount = request;
// Try to bump up small requests.
if (amount < IO_STREAM_BUF_SIZE) { amount = IO_STREAM_BUF_SIZE; }
// Don't read past EOF.
if (remaining_in_file < amount) { amount = remaining_in_file; }
// Only fill if the recalculated, possibly smaller request exceeds the
// amount available in the buffer.
if (amount > bytes_in_buf) {
S_fill(self, amount);
}
}
return self->buf;
}
void
InStream_advance_buf(InStream *self, char *buf) {
if (buf > self->limit) {
int64_t overrun = PTR_TO_I64(buf) - PTR_TO_I64(self->limit);
THROW(ERR, "Supplied value is %i64 bytes beyond end of buffer",
overrun);
}
else if (buf < self->buf) {
int64_t underrun = PTR_TO_I64(self->buf) - PTR_TO_I64(buf);
THROW(ERR, "Can't Advance_Buf backwards: (underrun: %i64))", underrun);
}
else {
self->buf = buf;
}
}
void
InStream_read_bytes(InStream *self, char* buf, size_t len) {
SI_read_bytes(self, buf, len);
}
static INLINE void
SI_read_bytes(InStream *self, char* buf, size_t len) {
const int64_t available = PTR_TO_I64(self->limit) - PTR_TO_I64(self->buf);
if (available >= (int64_t)len) {
// Request is entirely within buffer, so copy.
memcpy(buf, self->buf, len);
self->buf += len;
}
else {
// Pass along whatever we've got in the buffer.
if (available > 0) {
memcpy(buf, self->buf, (size_t)available);
buf += available;
len -= (size_t)available;
self->buf += available;
}
if (len < IO_STREAM_BUF_SIZE) {
// Ensure that we have enough mapped, then copy the rest.
int64_t got = S_refill(self);
if (got < (int64_t)len) {
int64_t orig_pos = SI_tell(self) - available;
int64_t orig_len = len + available;
THROW(ERR, "Read past EOF of %o (pos: %i64 len: %i64 "
"request: %i64)", self->filename, orig_pos,
self->len, orig_len);
}
memcpy(buf, self->buf, len);
self->buf += len;
}
else {
// Too big to handle via the buffer, so resort to a brute-force
// read.
const int64_t sub_file_pos = SI_tell(self);
const int64_t real_file_pos = sub_file_pos + self->offset;
bool_t success
= FH_Read(self->file_handle, buf, real_file_pos, len);
if (!success) {
RETHROW(INCREF(Err_get_error()));
}
InStream_seek(self, sub_file_pos + len);
}
}
}
int8_t
InStream_read_i8(InStream *self) {
return (int8_t)SI_read_u8(self);
}
static INLINE uint8_t
SI_read_u8(InStream *self) {
if (self->buf >= self->limit) { S_refill(self); }
return (uint8_t)(*self->buf++);
}
uint8_t
InStream_read_u8(InStream *self) {
return SI_read_u8(self);
}
static INLINE uint32_t
SI_read_u32(InStream *self) {
uint32_t retval;
SI_read_bytes(self, (char*)&retval, 4);
#ifdef LITTLE_END
retval = NumUtil_decode_bigend_u32((char*)&retval);
#endif
return retval;
}
uint32_t
InStream_read_u32(InStream *self) {
return SI_read_u32(self);
}
int32_t
InStream_read_i32(InStream *self) {
return (int32_t)SI_read_u32(self);
}
static INLINE uint64_t
SI_read_u64(InStream *self) {
uint64_t retval;
SI_read_bytes(self, (char*)&retval, 8);
#ifdef LITTLE_END
retval = NumUtil_decode_bigend_u64((char*)&retval);
#endif
return retval;
}
uint64_t
InStream_read_u64(InStream *self) {
return SI_read_u64(self);
}
int64_t
InStream_read_i64(InStream *self) {
return (int64_t)SI_read_u64(self);
}
float
InStream_read_f32(InStream *self) {
union { float f; uint32_t u32; } duo;
SI_read_bytes(self, (char*)&duo, sizeof(float));
#ifdef LITTLE_END
duo.u32 = NumUtil_decode_bigend_u32(&duo.u32);
#endif
return duo.f;
}
double
InStream_read_f64(InStream *self) {
union { double d; uint64_t u64; } duo;
SI_read_bytes(self, (char*)&duo, sizeof(double));
#ifdef LITTLE_END
duo.u64 = NumUtil_decode_bigend_u64(&duo.u64);
#endif
return duo.d;
}
uint32_t
InStream_read_c32(InStream *self) {
uint32_t retval = 0;
while (1) {
const uint8_t ubyte = SI_read_u8(self);
retval = (retval << 7) | (ubyte & 0x7f);
if ((ubyte & 0x80) == 0) {
break;
}
}
return retval;
}
uint64_t
InStream_read_c64(InStream *self) {
uint64_t retval = 0;
while (1) {
const uint8_t ubyte = SI_read_u8(self);
retval = (retval << 7) | (ubyte & 0x7f);
if ((ubyte & 0x80) == 0) {
break;
}
}
return retval;
}
int
InStream_read_raw_c64(InStream *self, char *buf) {
uint8_t *dest = (uint8_t*)buf;
do {
*dest = SI_read_u8(self);
} while ((*dest++ & 0x80) != 0);
return dest - (uint8_t*)buf;
}