/* 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
#include "Lucy/Util/ToolSet.h"
#include "charmony.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"
#include "Lucy/Util/NumberUtils.h"
// Inlined version of InStream_Tell.
static CFISH_INLINE int64_t
SI_tell(InStream *self);
// Inlined version of InStream_Read_Bytes.
static CFISH_INLINE void
SI_read_bytes(InStream *self, char* buf, int64_t len);
// Inlined version of InStream_Read_U8.
static CFISH_INLINE uint8_t
SI_read_u8(InStream *self, InStreamIVARS *const ivars);
// 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);
static CFISH_INLINE uint32_t
SI_read_cu32(InStream *self);
static CFISH_INLINE uint64_t
SI_read_cu64(InStream *self);
InStream*
InStream_open(Obj *file) {
InStream *self = (InStream*)Class_Make_Obj(INSTREAM);
return InStream_do_open(self, file);
}
InStream*
InStream_do_open(InStream *self, Obj *file) {
InStreamIVARS *const ivars = InStream_IVARS(self);
// Init.
ivars->buf = NULL;
ivars->limit = NULL;
ivars->offset = 0;
ivars->window = FileWindow_new();
// Obtain a FileHandle.
if (Obj_is_a(file, FILEHANDLE)) {
ivars->file_handle = (FileHandle*)INCREF(file);
}
else if (Obj_is_a(file, RAMFILE)) {
ivars->file_handle
= (FileHandle*)RAMFH_open(NULL, FH_READ_ONLY, (RAMFile*)file);
}
else if (Obj_is_a(file, STRING)) {
ivars->file_handle
= (FileHandle*)FSFH_open((String*)file, FH_READ_ONLY);
}
else {
Err_set_error(Err_new(Str_newf("Invalid type for param 'file': '%o'",
Obj_get_class_name(file))));
DECREF(self);
return NULL;
}
if (!ivars->file_handle) {
ERR_ADD_FRAME(Err_get_error());
DECREF(self);
return NULL;
}
// Get length and filename from the FileHandle.
ivars->filename = Str_Clone(FH_Get_Path(ivars->file_handle));
ivars->len = FH_Length(ivars->file_handle);
if (ivars->len == -1) {
ERR_ADD_FRAME(Err_get_error());
DECREF(self);
return NULL;
}
return self;
}
void
InStream_Close_IMP(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
if (ivars->file_handle) {
FH_Release_Window(ivars->file_handle, ivars->window);
// Note that we don't close the FileHandle, because it's probably
// shared.
DECREF(ivars->file_handle);
ivars->file_handle = NULL;
}
}
void
InStream_Destroy_IMP(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
if (ivars->file_handle) {
InStream_Close(self);
}
DECREF(ivars->filename);
DECREF(ivars->window);
SUPER_DESTROY(self, INSTREAM);
}
InStream*
InStream_Reopen_IMP(InStream *self, String *filename, int64_t offset,
int64_t len) {
InStreamIVARS *const ivars = InStream_IVARS(self);
if (!ivars->file_handle) {
THROW(ERR, "Can't Reopen() closed InStream %o", ivars->filename);
}
if (offset + len > FH_Length(ivars->file_handle)) {
THROW(ERR, "Offset + length too large (%i64 + %i64 > %i64)",
offset, len, FH_Length(ivars->file_handle));
}
Class *klass = InStream_get_class(self);
InStream *other = (InStream*)Class_Make_Obj(klass);
InStreamIVARS *const ovars = InStream_IVARS(other);
InStream_do_open(other, (Obj*)ivars->file_handle);
if (filename != NULL) {
String *temp = ovars->filename;
ovars->filename = Str_Clone(filename);
DECREF(temp);
}
ovars->offset = offset;
ovars->len = len;
InStream_Seek(other, 0);
return other;
}
InStream*
InStream_Clone_IMP(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
Class *klass = InStream_get_class(self);
InStream *twin = (InStream*)Class_Make_Obj(klass);
InStream_do_open(twin, (Obj*)ivars->file_handle);
InStream_Seek(twin, SI_tell(self));
return twin;
}
String*
InStream_Get_Filename_IMP(InStream *self) {
return InStream_IVARS(self)->filename;
}
static int64_t
S_refill(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
// Determine the amount to request.
const int64_t sub_file_pos = SI_tell(self);
const int64_t remaining = ivars->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)",
ivars->filename, ivars->offset, ivars->len);
}
// Make the request.
S_fill(self, amount);
return amount;
}
void
InStream_Refill_IMP(InStream *self) {
S_refill(self);
}
static void
S_fill(InStream *self, int64_t amount) {
InStreamIVARS *const ivars = InStream_IVARS(self);
FileWindow *const window = ivars->window;
const int64_t virtual_file_pos = SI_tell(self);
const int64_t real_file_pos = virtual_file_pos + ivars->offset;
const int64_t remaining = ivars->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: %i64 len: %i64 request: %i64)",
ivars->filename, virtual_file_pos, ivars->len, amount);
}
// Make the request.
if (FH_Window(ivars->file_handle, window, real_file_pos, amount)) {
char *fw_buf = FileWindow_Get_Buf(window);
int64_t fw_offset = FileWindow_Get_Offset(window);
int64_t fw_len = FileWindow_Get_Len(window);
char *const window_limit = fw_buf + fw_len;
ivars->buf = fw_buf
- fw_offset // theoretical start of real file
+ ivars->offset // top of virtual file
+ virtual_file_pos; // position within virtual file
ivars->limit = window_limit - ivars->buf > remaining
? ivars->buf + remaining
: window_limit;
}
else {
Err *error = Err_get_error();
String *str = Str_newf(" (%o)", ivars->filename);
Err_Cat_Mess(error, str);
DECREF(str);
RETHROW(INCREF(error));
}
}
void
InStream_Fill_IMP(InStream *self, int64_t amount) {
S_fill(self, amount);
}
void
InStream_Seek_IMP(InStream *self, int64_t target) {
InStreamIVARS *const ivars = InStream_IVARS(self);
FileWindow *const window = ivars->window;
char *fw_buf = FileWindow_Get_Buf(window);
int64_t fw_offset = FileWindow_Get_Offset(window);
int64_t fw_len = FileWindow_Get_Len(window);
int64_t virtual_window_top = fw_offset - ivars->offset;
int64_t virtual_window_end = virtual_window_top + fw_len;
if (target < 0) {
THROW(ERR, "Can't Seek '%o' to negative target %i64", ivars->filename,
target);
}
// Seek within window if possible.
else if (target >= virtual_window_top
&& target <= virtual_window_end
) {
ivars->buf = fw_buf - fw_offset + ivars->offset + target;
}
else if (target > ivars->len) {
THROW(ERR, "Can't Seek '%o' past EOF (%i64 > %i64)", ivars->filename,
target, ivars->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(ivars->file_handle, window);
ivars->buf = NULL;
ivars->limit = NULL;
FileWindow_Set_Offset(window, ivars->offset + target);
}
}
static CFISH_INLINE int64_t
SI_tell(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
char *fw_buf = FileWindow_Get_Buf(ivars->window);
int64_t pos_in_buf = CHY_PTR_TO_I64(ivars->buf) - CHY_PTR_TO_I64(fw_buf);
return pos_in_buf + FileWindow_Get_Offset(ivars->window) - ivars->offset;
}
int64_t
InStream_Tell_IMP(InStream *self) {
return SI_tell(self);
}
int64_t
InStream_Length_IMP(InStream *self) {
return InStream_IVARS(self)->len;
}
const char*
InStream_Buf_IMP(InStream *self, size_t request) {
InStreamIVARS *const ivars = InStream_IVARS(self);
const int64_t bytes_in_buf
= CHY_PTR_TO_I64(ivars->limit) - CHY_PTR_TO_I64(ivars->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 = ivars->len - SI_tell(self);
int64_t amount = (int64_t)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 ivars->buf;
}
void
InStream_Advance_Buf_IMP(InStream *self, const char *buf) {
InStreamIVARS *const ivars = InStream_IVARS(self);
if (buf > ivars->limit) {
int64_t overrun = CHY_PTR_TO_I64(buf) - CHY_PTR_TO_I64(ivars->limit);
THROW(ERR, "Supplied value is %i64 bytes beyond end of buffer",
overrun);
}
else if (buf < ivars->buf) {
int64_t underrun = CHY_PTR_TO_I64(ivars->buf) - CHY_PTR_TO_I64(buf);
THROW(ERR, "Can't Advance_Buf backwards: (underrun: %i64))", underrun);
}
else {
ivars->buf = buf;
}
}
void
InStream_Read_Bytes_IMP(InStream *self, char* buf, size_t len) {
#if SIZE_MAX > INT64_MAX
if (len >= INT64_MAX) {
THROW(ERR, "Can't read %u64 bytes", (uint64_t)len);
}
#endif
SI_read_bytes(self, buf, (int64_t)len);
}
static CFISH_INLINE void
SI_read_bytes(InStream *self, char* buf, int64_t len) {
InStreamIVARS *const ivars = InStream_IVARS(self);
const int64_t available
= CHY_PTR_TO_I64(ivars->limit) - CHY_PTR_TO_I64(ivars->buf);
if (available >= len) {
// Request is entirely within buffer, so copy.
memcpy(buf, ivars->buf, (size_t)len);
ivars->buf += len;
}
else {
// Pass along whatever we've got in the buffer.
if (available > 0) {
memcpy(buf, ivars->buf, (size_t)available);
buf += available;
len -= available;
ivars->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 < 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)", ivars->filename, orig_pos,
ivars->len, orig_len);
}
memcpy(buf, ivars->buf, (size_t)len);
ivars->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 + ivars->offset;
bool success
= FH_Read(ivars->file_handle, buf, real_file_pos, (size_t)len);
if (!success) {
RETHROW(INCREF(Err_get_error()));
}
InStream_Seek_IMP(self, sub_file_pos + len);
}
}
}
int8_t
InStream_Read_I8_IMP(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
return (int8_t)SI_read_u8(self, ivars);
}
static CFISH_INLINE uint8_t
SI_read_u8(InStream *self, InStreamIVARS *ivars) {
if (ivars->buf >= ivars->limit) { S_refill(self); }
return (uint8_t)(*ivars->buf++);
}
uint8_t
InStream_Read_U8_IMP(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
return SI_read_u8(self, ivars);
}
static CFISH_INLINE uint32_t
SI_read_u32(InStream *self) {
uint32_t retval;
SI_read_bytes(self, (char*)&retval, 4);
#ifdef CHY_LITTLE_END
retval = NumUtil_decode_bigend_u32((char*)&retval);
#endif
return retval;
}
uint32_t
InStream_Read_U32_IMP(InStream *self) {
return SI_read_u32(self);
}
int32_t
InStream_Read_I32_IMP(InStream *self) {
return (int32_t)SI_read_u32(self);
}
static CFISH_INLINE uint64_t
SI_read_u64(InStream *self) {
uint64_t retval;
SI_read_bytes(self, (char*)&retval, 8);
#ifdef CHY_LITTLE_END
retval = NumUtil_decode_bigend_u64((char*)&retval);
#endif
return retval;
}
uint64_t
InStream_Read_U64_IMP(InStream *self) {
return SI_read_u64(self);
}
int64_t
InStream_Read_I64_IMP(InStream *self) {
return (int64_t)SI_read_u64(self);
}
float
InStream_Read_F32_IMP(InStream *self) {
union { float f; uint32_t u32; } duo;
SI_read_bytes(self, (char*)&duo, sizeof(float));
#ifdef CHY_LITTLE_END
duo.u32 = NumUtil_decode_bigend_u32(&duo.u32);
#endif
return duo.f;
}
double
InStream_Read_F64_IMP(InStream *self) {
union { double d; uint64_t u64; } duo;
SI_read_bytes(self, (char*)&duo, sizeof(double));
#ifdef CHY_LITTLE_END
duo.u64 = NumUtil_decode_bigend_u64(&duo.u64);
#endif
return duo.d;
}
int32_t
InStream_Read_CI32_IMP(InStream *self) {
return (int32_t)SI_read_cu32(self);
}
uint32_t
InStream_Read_CU32_IMP(InStream *self) {
return SI_read_cu32(self);
}
static CFISH_INLINE uint32_t
SI_read_cu32(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
uint32_t retval = 0;
while (1) {
const uint8_t ubyte = SI_read_u8(self, ivars);
retval = (retval << 7) | (ubyte & 0x7f);
if ((ubyte & 0x80) == 0) {
break;
}
}
return retval;
}
int64_t
InStream_Read_CI64_IMP(InStream *self) {
return (int64_t)SI_read_cu64(self);
}
uint64_t
InStream_Read_CU64_IMP(InStream *self) {
return SI_read_cu64(self);
}
static CFISH_INLINE uint64_t
SI_read_cu64(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
uint64_t retval = 0;
while (1) {
const uint8_t ubyte = SI_read_u8(self, ivars);
retval = (retval << 7) | (ubyte & 0x7f);
if ((ubyte & 0x80) == 0) {
break;
}
}
return retval;
}
int
InStream_Read_Raw_C64_IMP(InStream *self, char *buf) {
InStreamIVARS *const ivars = InStream_IVARS(self);
uint8_t *dest = (uint8_t*)buf;
do {
*dest = SI_read_u8(self, ivars);
} while ((*dest++ & 0x80) != 0);
return (int)(dest - (uint8_t*)buf);
}
FileWindow*
InStream_Get_Window_IMP(InStream *self) {
return InStream_IVARS(self)->window;
}
FileHandle*
InStream_Get_Handle_IMP(InStream *self) {
return InStream_IVARS(self)->file_handle;
}
int64_t
InStream_Bytes_In_Buf_IMP(InStream *self) {
InStreamIVARS *const ivars = InStream_IVARS(self);
return (int64_t)(ivars->limit - ivars->buf);
}