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
|
/*
* Copyright 2023 jacqueline <me@jacqueline.id.au>
*
* SPDX-License-Identifier: GPL-3.0-only
*/
#include "readahead_source.hpp"
#include <cstddef>
#include <cstdint>
#include <memory>
#include "esp_heap_caps.h"
#include "esp_log.h"
#include "ff.h"
#include "audio_source.hpp"
#include "codec.hpp"
#include "freertos/portmacro.h"
#include "idf_additions.h"
#include "spi.hpp"
#include "tasks.hpp"
#include "types.hpp"
namespace audio {
static constexpr char kTag[] = "readahead";
static constexpr size_t kBufferSize = 1024 * 512;
ReadaheadSource::ReadaheadSource(tasks::Worker& worker,
std::unique_ptr<codecs::IStream> wrapped)
: IStream(wrapped->type()),
worker_(worker),
wrapped_(std::move(wrapped)),
is_refilling_(false),
buffer_(xStreamBufferCreateWithCaps(kBufferSize, 1, MALLOC_CAP_SPIRAM)),
tell_(wrapped_->CurrentPosition()) {}
ReadaheadSource::~ReadaheadSource() {
is_refilling_.wait(true);
vStreamBufferDeleteWithCaps(buffer_);
}
auto ReadaheadSource::Read(cpp::span<std::byte> dest) -> ssize_t {
size_t bytes_written = 0;
// Fill the destination from our buffer, until either the buffer is drained
// or the destination is full.
while (!dest.empty() && (is_refilling_ || !xStreamBufferIsEmpty(buffer_))) {
size_t bytes_read = xStreamBufferReceive(buffer_, dest.data(),
dest.size_bytes(), portMAX_DELAY);
tell_ += bytes_read;
bytes_written += bytes_read;
dest = dest.subspan(bytes_read);
}
// After the loop, we've either written everything that was asked for, or
// we're out of data.
if (!dest.empty()) {
// Out of data in the buffer. Finish using the wrapped stream.
size_t extra_bytes = wrapped_->Read(dest);
tell_ += extra_bytes;
bytes_written += extra_bytes;
// Check for EOF in the wrapped stream.
if (extra_bytes < dest.size_bytes()) {
return bytes_written;
}
}
// After this point, we're done writing to `dest`. It's either empty, or the
// underlying source is EOF.
// If we're here, then there is more data to be read from the wrapped stream.
// Ensure the readahead is running.
if (!is_refilling_ &&
xStreamBufferBytesAvailable(buffer_) < kBufferSize / 4) {
is_refilling_ = true;
std::function<void(void)> refill = [this]() {
// Try to keep larger than most reasonable FAT sector sizes for more
// efficient disk reads.
constexpr size_t kMaxSingleRead = 1024 * 16;
std::byte working_buf[kMaxSingleRead];
for (;;) {
size_t bytes_to_read = std::min<size_t>(
kMaxSingleRead, xStreamBufferSpacesAvailable(buffer_));
if (bytes_to_read == 0) {
break;
}
size_t read = wrapped_->Read({working_buf, bytes_to_read});
if (read > 0) {
xStreamBufferSend(buffer_, working_buf, read, 0);
}
if (read < bytes_to_read) {
break;
}
}
is_refilling_ = false;
is_refilling_.notify_all();
};
worker_.Dispatch(refill);
}
return bytes_written;
}
auto ReadaheadSource::CanSeek() -> bool {
return wrapped_->CanSeek();
}
auto ReadaheadSource::SeekTo(int64_t destination, SeekFrom from) -> void {
// Seeking blows away all of our prefetched data. To do this safely, we
// first need to wait for the refill task to finish.
ESP_LOGI(kTag, "dropping readahead due to seek");
is_refilling_.wait(true);
// It's now safe to clear out the buffer.
xStreamBufferReset(buffer_);
wrapped_->SeekTo(destination, from);
// Make sure our tell is up to date with the new location.
tell_ = wrapped_->CurrentPosition();
}
auto ReadaheadSource::CurrentPosition() -> int64_t {
return tell_;
}
} // namespace audio
|
