src/mgr/StoreToCommWriter.cc

   1 /*
   2  * Copyright (C) 1996-2020 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/TextException.h"
  13 #include "comm/Connection.h"
  14 #include "comm/Write.h"
  15 #include "CommCalls.h"
  16 #include "ipc/FdNotes.h"
  17 #include "mgr/StoreToCommWriter.h"
  18 #include "Store.h"
  19 #include "StoreClient.h"
  20
  21 CBDATA_NAMESPACED_CLASS_INIT(Mgr, StoreToCommWriter);
  22
  23 Mgr::StoreToCommWriter::StoreToCommWriter(const Comm::ConnectionPointer &conn, StoreEntry* anEntry):
  24     AsyncJob("Mgr::StoreToCommWriter"),
  25     clientConnection(conn), entry(anEntry), sc(NULL), writeOffset(0), closer(NULL)
  26 {
  27     debugs(16, 6, HERE << 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, HERE);
  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 != NULL) {
  47             comm_remove_close_handler(clientConnection->fd, closer);
  48             closer = NULL;
  49         }
  50         clientConnection->close();
  51     }
  52 }
  53
  54 void
  55 Mgr::StoreToCommWriter::start()
  56 {
  57     debugs(16, 6, HERE);
  58     Must(Comm::IsConnOpen(clientConnection));
  59     Must(entry != NULL);
  60     entry->registerAbort(&StoreToCommWriter::Abort, this);
  61     sc = storeClientListAdd(entry, this);
  62     Must(sc != NULL);
  63
  64     // initiate the receive-from-store, write-to-comm sequence
  65     scheduleStoreCopy();
  66 }
  67
  68 void
  69 Mgr::StoreToCommWriter::scheduleStoreCopy()
  70 {
  71     debugs(16, 6, HERE);
  72     Must(entry != NULL);
  73     Must(sc != NULL);
  74     StoreIOBuffer readBuf(sizeof(buffer), writeOffset, buffer);
  75     storeClientCopy(sc, entry, readBuf, &NoteStoreCopied, this);
  76 }
  77
  78 void
  79 Mgr::StoreToCommWriter::NoteStoreCopied(void* data, StoreIOBuffer ioBuf)
  80 {
  81     Must(data != NULL);
  82     // make sync Store call async to get async call protections and features
  83     StoreToCommWriter* writer = static_cast<StoreToCommWriter*>(data);
  84     typedef UnaryMemFunT<StoreToCommWriter, StoreIOBuffer> MyDialer;
  85     AsyncCall::Pointer call =
  86         asyncCall(16, 5, "Mgr::StoreToCommWriter::noteStoreCopied",
  87                   MyDialer(writer, &StoreToCommWriter::noteStoreCopied, ioBuf));
  88     ScheduleCallHere(call);
  89 }
  90
  91 void
  92 Mgr::StoreToCommWriter::noteStoreCopied(StoreIOBuffer ioBuf)
  93 {
  94     debugs(16, 6, HERE);
  95     Must(!ioBuf.flags.error);
  96     if (ioBuf.length > 0)
  97         scheduleCommWrite(ioBuf); // write received action results to client
  98     else
  99         Must(doneAll()); // otherwise, why would Store call us with no data?
 100 }
 101
 102 void
 103 Mgr::StoreToCommWriter::scheduleCommWrite(const StoreIOBuffer& ioBuf)
 104 {
 105     debugs(16, 6, HERE);
 106     Must(Comm::IsConnOpen(clientConnection));
 107     Must(ioBuf.data != NULL);
 108     // write filled buffer
 109     typedef CommCbMemFunT<StoreToCommWriter, CommIoCbParams> MyDialer;
 110     AsyncCall::Pointer writer =
 111         asyncCall(16, 5, "Mgr::StoreToCommWriter::noteCommWrote",
 112                   MyDialer(this, &StoreToCommWriter::noteCommWrote));
 113     Comm::Write(clientConnection, ioBuf.data, ioBuf.length, writer, NULL);
 114 }
 115
 116 void
 117 Mgr::StoreToCommWriter::noteCommWrote(const CommIoCbParams& params)
 118 {
 119     debugs(16, 6, HERE);
 120     Must(params.flag == Comm::OK);
 121     Must(clientConnection != NULL && params.fd == clientConnection->fd);
 122     Must(params.size != 0);
 123     writeOffset += params.size;
 124     if (!doneAll())
 125         scheduleStoreCopy(); // retrieve the next data portion
 126 }
 127
 128 void
 129 Mgr::StoreToCommWriter::noteCommClosed(const CommCloseCbParams &)
 130 {
 131     debugs(16, 6, HERE);
 132     Must(!Comm::IsConnOpen(clientConnection));
 133     mustStop("commClosed");
 134 }
 135
 136 void
 137 Mgr::StoreToCommWriter::swanSong()
 138 {
 139     debugs(16, 6, HERE);
 140     if (entry != NULL) {
 141         if (sc != NULL) {
 142             storeUnregister(sc, entry, this);
 143             sc = NULL;
 144         }
 145         entry->unregisterAbort();
 146         entry->unlock("Mgr::StoreToCommWriter::swanSong");
 147         entry = NULL;
 148     }
 149     close();
 150 }
 151
 152 bool
 153 Mgr::StoreToCommWriter::doneAll() const
 154 {
 155     return entry &&
 156            entry->store_status == STORE_OK && // the action is over
 157            writeOffset >= entry->objectLen(); // we wrote all the results
 158 }
 159
 160 void
 161 Mgr::StoreToCommWriter::Abort(void* param)
 162 {
 163     StoreToCommWriter* mgrWriter = static_cast<StoreToCommWriter*>(param);
 164     if (Comm::IsConnOpen(mgrWriter->clientConnection))
 165         mgrWriter->clientConnection->close();
 166 }
 167