src/mgr/StoreToCommWriter.cc

   1 /*
   2  * Copyright (C) 1996-2023 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 16    Cache Manager API */
  10
  11 #include "squid.h"
  12 #include "base/AsyncCbdataCalls.h"
  13 #include "base/TextException.h"
  14 #include "comm.h"
  15 #include "comm/Connection.h"
  16 #include "comm/Write.h"
  17 #include "CommCalls.h"
  18 #include "ipc/FdNotes.h"
  19 #include "mgr/StoreToCommWriter.h"
  20 #include "Store.h"
  21 #include "StoreClient.h"
  22
  23 Mgr::StoreToCommWriter::StoreToCommWriter(const Comm::ConnectionPointer &conn, StoreEntry* anEntry):
  24     AsyncJob("Mgr::StoreToCommWriter"),
  25     clientConnection(conn), entry(anEntry), sc(nullptr), writeOffset(0), closer(nullptr)
  26 {
  27     debugs(16, 6, clientConnection);
  28     closer = asyncCall(16, 5, "Mgr::StoreToCommWriter::noteCommClosed",
  29                        CommCbMemFunT<StoreToCommWriter, CommCloseCbParams>(this, &StoreToCommWriter::noteCommClosed));
  30     comm_add_close_handler(clientConnection->fd, closer);
  31 }
  32
  33 Mgr::StoreToCommWriter::~StoreToCommWriter()
  34 {
  35     debugs(16, 6, MYNAME);
  36     assert(!entry);
  37     assert(!sc);
  38     close();
  39 }
  40
  41 /// closes our copy of the client HTTP connection socket
  42 void
  43 Mgr::StoreToCommWriter::close()
  44 {
  45     if (Comm::IsConnOpen(clientConnection)) {
  46         if (closer != nullptr) {
  47             comm_remove_close_handler(clientConnection->fd, closer);
  48             closer = nullptr;
  49         }
  50         clientConnection->close();
  51     }
  52 }
  53
  54 void
  55 Mgr::StoreToCommWriter::start()
  56 {
  57     debugs(16, 6, MYNAME);
  58     Must(Comm::IsConnOpen(clientConnection));
  59     Must(entry != nullptr);
  60     AsyncCall::Pointer call = asyncCall(16, 4, "StoreToCommWriter::Abort", cbdataDialer(&StoreToCommWriter::HandleStoreAbort, this));
  61     entry->registerAbortCallback(call);
  62     sc = storeClientListAdd(entry, this);
  63     Must(sc != nullptr);
  64
  65     // initiate the receive-from-store, write-to-comm sequence
  66     scheduleStoreCopy();
  67 }
  68
  69 void
  70 Mgr::StoreToCommWriter::scheduleStoreCopy()
  71 {
  72     debugs(16, 6, MYNAME);
  73     Must(entry != nullptr);
  74     Must(sc != nullptr);
  75     StoreIOBuffer readBuf(sizeof(buffer), writeOffset, buffer);
  76     storeClientCopy(sc, entry, readBuf, &NoteStoreCopied, this);
  77 }
  78
  79 void
  80 Mgr::StoreToCommWriter::NoteStoreCopied(void* data, StoreIOBuffer ioBuf)
  81 {
  82     Must(data != nullptr);
  83     // make sync Store call async to get async call protections and features
  84     StoreToCommWriter* writer = static_cast<StoreToCommWriter*>(data);
  85     typedef UnaryMemFunT<StoreToCommWriter, StoreIOBuffer> MyDialer;
  86     AsyncCall::Pointer call =
  87         asyncCall(16, 5, "Mgr::StoreToCommWriter::noteStoreCopied",
  88                   MyDialer(writer, &StoreToCommWriter::noteStoreCopied, ioBuf));
  89     ScheduleCallHere(call);
  90 }
  91
  92 void
  93 Mgr::StoreToCommWriter::noteStoreCopied(StoreIOBuffer ioBuf)
  94 {
  95     debugs(16, 6, MYNAME);
  96     Must(!ioBuf.flags.error);
  97     if (ioBuf.length > 0)
  98         scheduleCommWrite(ioBuf); // write received action results to client
  99     else
 100         Must(doneAll()); // otherwise, why would Store call us with no data?
 101 }
 102
 103 void
 104 Mgr::StoreToCommWriter::scheduleCommWrite(const StoreIOBuffer& ioBuf)
 105 {
 106     debugs(16, 6, MYNAME);
 107     Must(Comm::IsConnOpen(clientConnection));
 108     Must(ioBuf.data != nullptr);
 109     // write filled buffer
 110     typedef CommCbMemFunT<StoreToCommWriter, CommIoCbParams> MyDialer;
 111     AsyncCall::Pointer writer =
 112         asyncCall(16, 5, "Mgr::StoreToCommWriter::noteCommWrote",
 113                   MyDialer(this, &StoreToCommWriter::noteCommWrote));
 114     Comm::Write(clientConnection, ioBuf.data, ioBuf.length, writer, nullptr);
 115 }
 116
 117 void
 118 Mgr::StoreToCommWriter::noteCommWrote(const CommIoCbParams& params)
 119 {
 120     debugs(16, 6, MYNAME);
 121     Must(params.flag == Comm::OK);
 122     Must(clientConnection != nullptr && params.fd == clientConnection->fd);
 123     Must(params.size != 0);
 124     writeOffset += params.size;
 125     if (!doneAll())
 126         scheduleStoreCopy(); // retrieve the next data portion
 127 }
 128
 129 void
 130 Mgr::StoreToCommWriter::noteCommClosed(const CommCloseCbParams &)
 131 {
 132     debugs(16, 6, MYNAME);
 133     if (clientConnection) {
 134         clientConnection->noteClosure();
 135         clientConnection = nullptr;
 136     }
 137     closer = nullptr;
 138     mustStop("commClosed");
 139 }
 140
 141 void
 142 Mgr::StoreToCommWriter::swanSong()
 143 {
 144     debugs(16, 6, MYNAME);
 145     if (entry != nullptr) {
 146         if (sc != nullptr) {
 147             storeUnregister(sc, entry, this);
 148             sc = nullptr;
 149         }
 150         entry->unregisterAbortCallback("StoreToCommWriter done");
 151         entry->unlock("Mgr::StoreToCommWriter::swanSong");
 152         entry = nullptr;
 153     }
 154     close();
 155 }
 156
 157 bool
 158 Mgr::StoreToCommWriter::doneAll() const
 159 {
 160     return entry &&
 161            entry->store_status == STORE_OK && // the action is over
 162            writeOffset >= entry->objectLen(); // we wrote all the results
 163 }
 164
 165 void
 166 Mgr::StoreToCommWriter::HandleStoreAbort(StoreToCommWriter *mgrWriter)
 167 {
 168     if (Comm::IsConnOpen(mgrWriter->clientConnection))
 169         mgrWriter->clientConnection->close();
 170 }
 171