1+ /* -------------------------------------------------------------------------
2+ *
3+ * Licensed to the Apache Software Foundation (ASF) under one
4+ * or more contributor license agreements. See the NOTICE file
5+ * distributed with this work for additional information
6+ * regarding copyright ownership. The ASF licenses this file
7+ * to you under the Apache License, Version 2.0 (the
8+ * "License"); you may not use this file except in compliance
9+ * with the License. You may obtain a copy of the License at
10+ *
11+ * http://www.apache.org/licenses/LICENSE-2.0
12+ *
13+ * Unless required by applicable law or agreed to in writing,
14+ * software distributed under the License is distributed on an
15+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+ * KIND, either express or implied. See the License for the
17+ * specific language governing permissions and limitations
18+ * under the License.
19+ *
20+ * fast_io.cc
21+ *
22+ * IDENTIFICATION
23+ * contrib/pax_storage/src/cpp/comm/fast_io.cc
24+ *
25+ *-------------------------------------------------------------------------
26+ */
27+
28+ #include " fast_io.h"
29+
30+ namespace pax
31+ {
32+
33+ bool IOUringFastIO::available () {
34+ static char support_io_uring = 0 ;
35+
36+ if (support_io_uring == 1 ) return true ;
37+ if (support_io_uring == -1 ) return false ;
38+
39+ struct io_uring ring;
40+ bool supported = (io_uring_queue_init (128 , &ring, 0 ) == 0 );
41+ if (supported) {
42+ io_uring_queue_exit (&ring);
43+ }
44+ support_io_uring = supported ? 1 : -1 ;
45+ return supported;
46+ }
47+
48+ // if pair.first == 0, all read requests are successful
49+ // pair.second indicates the number of successful read requests
50+ std::pair<int , int > IOUringFastIO::read (int fd, std::vector<IORequest> &request, std::vector<bool > &result) {
51+ size_t index = 0 ;
52+ int success_read = 0 ;
53+ int retcode = 0 ;
54+ size_t completed = 0 ;
55+ size_t total_requests = request.size ();
56+
57+ // Implementation for synchronous read using io_uring
58+ if (uring_likely (request.empty ())) return {0 , 0 };
59+ if (status_ != ' i' return {-EINVAL, 0 };
60+
61+ result.resize (request.size (), false );
62+
63+ while (completed < total_requests) {
64+ struct io_uring_sqe *sqe;
65+ struct io_uring_cqe *cqe;
66+ unsigned head;
67+ unsigned count;
68+ int rc;
69+ // Submit read requests
70+ while (index < total_requests) {
71+ sqe = io_uring_get_sqe (&ring_);
72+ if (!sqe) break ; // No more SQEs available, retry later
73+
74+ io_uring_prep_read (sqe, fd, request[index].buffer , request[index].size , request[index].offset );
75+ io_uring_sqe_set_data (sqe, (void *)(uintptr_t )index);
76+ index++;
77+ }
78+
79+ // submit and wait for completions
80+ do {
81+ rc = io_uring_submit_and_wait (&ring_, 1 );
82+ } while (rc == -EINTR);
83+ if (rc < 0 ) return {rc, success_read};
84+
85+ count = 0 ;
86+ io_uring_for_each_cqe (&ring_, head, cqe) {
87+ size_t req_index = (size_t )(uintptr_t )io_uring_cqe_get_data (cqe);
88+ if (cqe->res >= 0 ) {
89+ // Successful read
90+ result[req_index] = true ;
91+ success_read++;
92+ } else if (retcode == 0 ) {
93+ retcode = cqe->res ; // capture the first error
94+ }
95+ completed++;
96+ count++;
97+ }
98+ io_uring_cq_advance (&ring_, count);
99+ }
100+ return {retcode, success_read}; // Placeholder
101+ }
102+
103+ std::pair<int , int > SyncFastIO::read (int fd, std::vector<IORequest> &request, std::vector<bool > &result) {
104+ size_t total_requests = request.size ();
105+ if (total_requests == 0 ) return {0 , 0 };
106+
107+ result.resize (total_requests, false );
108+
109+ int success_read = 0 ;
110+ int retcode = 0 ;
111+
112+ for (size_t i = 0 ; i < total_requests; ++i) {
113+ ssize_t bytes_read = 0 ;
114+ ssize_t nbytes;
115+ auto &req = request[i];
116+ do {
117+ nbytes = pread (fd, (char *)req.buffer + bytes_read, req.size - bytes_read, req.offset + bytes_read);
118+ if (nbytes > 0 ) bytes_read += nbytes;
119+ } while ((nbytes == -1 && errno == EINTR) || (nbytes > 0 && static_cast <size_t >(bytes_read) < req.size ));
120+
121+ if (bytes_read < 0 ) {
122+ if (retcode == 0 ) {
123+ retcode = static_cast <int >(bytes_read); // capture first error
124+ }
125+ } else if (static_cast <size_t >(bytes_read) == request[i].size ) {
126+ result[i] = true ;
127+ success_read++;
128+ }
129+ }
130+
131+ return {retcode, success_read};
132+ }
133+
134+ } // namespace pax
0 commit comments