CollapsedForwarding.cc
Go to the documentation of this file.
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 17 Request Forwarding */
10 
11 #include "squid.h"
12 #include "CollapsedForwarding.h"
13 #include "globals.h"
14 #include "ipc/mem/Segment.h"
15 #include "ipc/Messages.h"
16 #include "ipc/Port.h"
17 #include "ipc/TypedMsgHdr.h"
18 #include "MemObject.h"
19 #include "SquidConfig.h"
20 #include "Store.h"
21 #include "store_key_md5.h"
22 #include "tools.h"
23 
25 static const char *const ShmLabel = "cf";
27 // TODO: make configurable or compute from squid.conf settings if possible
28 static const int QueueCapacity = 1024;
29 
30 std::unique_ptr<CollapsedForwarding::Queue> CollapsedForwarding::queue;
31 
34 {
35 public:
37 
39  void stat(std::ostream &);
40 
41 public:
42  int sender;
43 
46 };
47 
48 void
50 {
51  os << "sender: " << sender << ", xitIndex: " << xitIndex;
52 }
53 
54 // CollapsedForwarding
55 
56 void
58 {
59  Must(!queue.get());
60  if (UsingSmp() && IamWorkerProcess())
61  queue.reset(new Queue(ShmLabel, KidIdentifier));
62 }
63 
64 void
65 CollapsedForwarding::Broadcast(const StoreEntry &e, const bool includingThisWorker)
66 {
67  if (!e.hasTransients() ||
69  debugs(17, 7, "nobody reads " << e);
70  return;
71  }
72 
73  debugs(17, 5, e);
74  Broadcast(e.mem_obj->xitTable.index, includingThisWorker);
75 }
76 
77 void
78 CollapsedForwarding::Broadcast(const sfileno index, const bool includingThisWorker)
79 {
80  if (!queue.get())
81  return;
82 
84  msg.sender = KidIdentifier;
85  msg.xitIndex = index;
86 
87  debugs(17, 7, "entry " << index << " to " << Config.workers << (includingThisWorker ? "" : "-1") << " workers");
88 
89  // TODO: send only to workers who are waiting for data
90  for (int workerId = 1; workerId <= Config.workers; ++workerId) {
91  try {
92  if ((workerId != KidIdentifier || includingThisWorker) && queue->push(workerId, msg))
93  Notify(workerId);
94  } catch (const Queue::Full &) {
95  debugs(17, DBG_IMPORTANT, "ERROR: Collapsed forwarding " <<
96  "queue overflow for kid" << workerId <<
97  " at " << queue->outSize(workerId) << " items");
98  // TODO: grow queue size
99  }
100  }
101 }
102 
103 void
104 CollapsedForwarding::Notify(const int workerId)
105 {
106  // TODO: Count and report the total number of notifications, pops, pushes.
107  debugs(17, 7, "to kid" << workerId);
108  Ipc::TypedMsgHdr msg;
110  msg.putInt(KidIdentifier);
111  const String addr = Ipc::Port::MakeAddr(Ipc::strandAddrLabel, workerId);
112  Ipc::SendMessage(addr, msg);
113 }
114 
115 void
116 CollapsedForwarding::HandleNewData(const char *const when)
117 {
118  debugs(17, 4, "popping all " << when);
120  int workerId;
121  int poppedCount = 0;
122  while (queue->pop(workerId, msg)) {
123  debugs(17, 3, "message from kid" << workerId);
124  if (workerId != msg.sender) {
125  debugs(17, DBG_IMPORTANT, "mismatching kid IDs: " << workerId <<
126  " != " << msg.sender);
127  }
128 
129  debugs(17, 7, "handling entry " << msg.xitIndex << " in transients_map");
131  debugs(17, 7, "handled entry " << msg.xitIndex << " in transients_map");
132 
133  // XXX: stop and schedule an async call to continue
134  ++poppedCount;
135  assert(poppedCount < SQUID_MAXFD);
136  }
137 }
138 
139 void
141 {
142  const int from = msg.getInt();
143  debugs(17, 7, "from " << from);
144  assert(queue.get());
145  queue->clearReaderSignal(from);
146  HandleNewData("after notification");
147 }
148 
149 void
151 {
152  if (queue.get()) {
153  os << "Transients queues:\n";
154  queue->stat<CollapsedForwardingMsg>(os);
155  }
156 }
157 
160 {
161 public:
162  /* RegisteredRunner API */
164  virtual ~CollapsedForwardingRr();
165 
166 protected:
167  virtual void create();
168  virtual void open();
169 
170 private:
172 };
173 
175 
177 {
178  Must(!owner);
180  sizeof(CollapsedForwardingMsg),
181  QueueCapacity);
182 }
183 
185 {
187 }
188 
190 {
191  delete owner;
192 }
193 
static String MakeAddr(const char *proccessLabel, int id)
calculates IPC message address for strand id of processLabel type
Definition: Port.cc:51
int transientReaders(const StoreEntry &) const
number of the transient entry readers some time ago
Definition: Controller.cc:658
sfileno xitIndex
transients index, so that workers can find [private] entries to sync
void setType(int aType)
sets message type; use MessageType enum
Definition: TypedMsgHdr.cc:107
MemObject * mem_obj
Definition: Store.h:213
int KidIdentifier
const char strandAddrLabel[]
strand's listening address unique label
Definition: Port.cc:22
bool IamWorkerProcess()
whether the current process handles HTTP transactions and such
Definition: stub_tools.cc:49
virtual void create()
called when the runner should create a new memory segment
#define DBG_IMPORTANT
Definition: Debug.h:46
static void HandleNewData(const char *const when)
handle new data messages in IPC queue
static std::unique_ptr< Queue > queue
IPC queue.
void stat(std::ostream &)
prints message parameters; suitable for cache manager reports
static Owner * Init(const String &id, const int processCount, const int processIdOffset, const unsigned int maxItemSize, const int capacity)
Definition: Queue.cc:361
void putInt(int n)
store an integer
Definition: TypedMsgHdr.cc:126
#define NULL
Definition: types.h:166
#define debugs(SECTION, LEVEL, CONTENT)
Definition: Debug.h:128
static void HandleNotification(const Ipc::TypedMsgHdr &msg)
handle queue push notifications from worker or disker
int32_t index
entry position inside the in-transit table
Definition: MemObject.h:172
XitTable xitTable
current [shared] memory caching state for the entry
Definition: MemObject.h:175
static void StatQueue(std::ostream &)
prints IPC message queue state; suitable for cache manager reports
RunnerRegistrationEntry(CollapsedForwardingRr)
initializes shared queue used by CollapsedForwarding
void SendMessage(const String &toAddress, const TypedMsgHdr &message)
Definition: UdsOp.cc:188
#define assert(EX)
Definition: assert.h:19
int getInt() const
load an integer
Definition: TypedMsgHdr.cc:118
void syncCollapsed(const sfileno)
Update local intransit entry after changes made by appending worker.
Definition: Controller.cc:795
static const int QueueCapacity
a single worker-to-worker queue capacity
signed_int32_t sfileno
Definition: forward.h:22
@ mtCollapsedForwardingNotification
Definition: Messages.h:24
Ipc::MultiQueue::Owner * owner
#define Must(condition)
Like assert() but throws an exception instead of aborting the process.
Definition: TextException.h:69
struct msghdr with a known type, fixed-size I/O and control buffers
Definition: TypedMsgHdr.h:31
bool hasTransients() const
whether there is a corresponding locked transients table entry
Definition: Store.h:203
static void Notify(const int workerId)
kick worker with empty IPC queue
static void Init()
open shared memory segment
static void Broadcast(const StoreEntry &e, const bool includingThisWorker=false)
notify other workers about changes in entry state (e.g., new data)
bool UsingSmp()
Whether there should be more than one worker process running.
Definition: tools.cc:658
static const char *const ShmLabel
shared memory segment path to use for CollapsedForwarding queue
class SquidConfig Config
Definition: SquidConfig.cc:12
int sender
kid ID of sending process
Controller & Root()
safely access controller singleton
Definition: Controller.cc:938

 

Introduction

Documentation

Support

Miscellaneous

Web Site Translations

Mirrors