PriorityChansConverger (empty) → 0.1
raw patch · 18 files changed
+2691/−0 lines, 18 filesdep +basedep +containersdep +stmsetup-changed
Dependencies added: base, containers, stm
Files
- COPYING +502/−0
- COPYRIGHT +18/−0
- Control/Concurrent/ConcurrentUISupport.hs +227/−0
- Control/Concurrent/PriorityChansConverger.hs +20/−0
- Control/Concurrent/PriorityChansConverger/Commons.hs +26/−0
- Control/Concurrent/PriorityChansConverger/PriorityChansConverger.hs +284/−0
- Control/Concurrent/PriorityChansConverger/PriorityChansConvergerSTM.hs +428/−0
- Control/Concurrent/STM/TChan/ExtsCommons.hs +22/−0
- Control/Concurrent/STM/TChan/TChanB.hs +18/−0
- Control/Concurrent/STM/TChan/TChanB_.hs +127/−0
- Control/Concurrent/STM/TChan/TChanL.hs +19/−0
- Control/Concurrent/STM/TChan/TChanL_.hs +88/−0
- Data/MyHelpers.hs +18/−0
- Data/MyHelpers_.hs +384/−0
- PriorityChansConverger.cabal +56/−0
- Setup.hs +4/−0
- doinst.sh +8/−0
- examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs +442/−0
+ COPYING view
@@ -0,0 +1,502 @@+ GNU LESSER GENERAL PUBLIC LICENSE+ Version 2.1, February 1999++ Copyright (C) 1991, 1999 Free Software Foundation, Inc.+ 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA+ Everyone is permitted to copy and distribute verbatim copies+ of this license document, but changing it is not allowed.++[This is the first released version of the Lesser GPL. It also counts+ as the successor of the GNU Library Public License, version 2, hence+ the version number 2.1.]++ Preamble++ The licenses for most software are designed to take away your+freedom to share and change it. By contrast, the GNU General Public+Licenses are intended to guarantee your freedom to share and change+free software--to make sure the software is free for all its users.++ This license, the Lesser General Public License, applies to some+specially designated software packages--typically libraries--of the+Free Software Foundation and other authors who decide to use it. You+can use it too, but we suggest you first think carefully about whether+this license or the ordinary General Public License is the better+strategy to use in any particular case, based on the explanations below.++ When we speak of free software, we are referring to freedom of use,+not price. Our General Public Licenses are designed to make sure that+you have the freedom to distribute copies of free software (and charge+for this service if you wish); that you receive source code or can get+it if you want it; that you can change the software and use pieces of+it in new free programs; and that you are informed that you can do+these things.++ To protect your rights, we need to make restrictions that forbid+distributors to deny you these rights or to ask you to surrender these+rights. These restrictions translate to certain responsibilities for+you if you distribute copies of the library or if you modify it.++ For example, if you distribute copies of the library, whether gratis+or for a fee, you must give the recipients all the rights that we gave+you. You must make sure that they, too, receive or can get the source+code. If you link other code with the library, you must provide+complete object files to the recipients, so that they can relink them+with the library after making changes to the library and recompiling+it. And you must show them these terms so they know their rights.++ We protect your rights with a two-step method: (1) we copyright the+library, and (2) we offer you this license, which gives you legal+permission to copy, distribute and/or modify the library.++ To protect each distributor, we want to make it very clear that+there is no warranty for the free library. Also, if the library is+modified by someone else and passed on, the recipients should know+that what they have is not the original version, so that the original+author's reputation will not be affected by problems that might be+introduced by others.++ Finally, software patents pose a constant threat to the existence of+any free program. We wish to make sure that a company cannot+effectively restrict the users of a free program by obtaining a+restrictive license from a patent holder. Therefore, we insist that+any patent license obtained for a version of the library must be+consistent with the full freedom of use specified in this license.++ Most GNU software, including some libraries, is covered by the+ordinary GNU General Public License. This license, the GNU Lesser+General Public License, applies to certain designated libraries, and+is quite different from the ordinary General Public License. We use+this license for certain libraries in order to permit linking those+libraries into non-free programs.++ When a program is linked with a library, whether statically or using+a shared library, the combination of the two is legally speaking a+combined work, a derivative of the original library. The ordinary+General Public License therefore permits such linking only if the+entire combination fits its criteria of freedom. The Lesser General+Public License permits more lax criteria for linking other code with+the library.++ We call this license the "Lesser" General Public License because it+does Less to protect the user's freedom than the ordinary General+Public License. It also provides other free software developers Less+of an advantage over competing non-free programs. These disadvantages+are the reason we use the ordinary General Public License for many+libraries. However, the Lesser license provides advantages in certain+special circumstances.++ For example, on rare occasions, there may be a special need to+encourage the widest possible use of a certain library, so that it becomes+a de-facto standard. To achieve this, non-free programs must be+allowed to use the library. A more frequent case is that a free+library does the same job as widely used non-free libraries. In this+case, there is little to gain by limiting the free library to free+software only, so we use the Lesser General Public License.++ In other cases, permission to use a particular library in non-free+programs enables a greater number of people to use a large body of+free software. For example, permission to use the GNU C Library in+non-free programs enables many more people to use the whole GNU+operating system, as well as its variant, the GNU/Linux operating+system.++ Although the Lesser General Public License is Less protective of the+users' freedom, it does ensure that the user of a program that is+linked with the Library has the freedom and the wherewithal to run+that program using a modified version of the Library.++ The precise terms and conditions for copying, distribution and+modification follow. Pay close attention to the difference between a+"work based on the library" and a "work that uses the library". The+former contains code derived from the library, whereas the latter must+be combined with the library in order to run.++ GNU LESSER GENERAL PUBLIC LICENSE+ TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION++ 0. This License Agreement applies to any software library or other+program which contains a notice placed by the copyright holder or+other authorized party saying it may be distributed under the terms of+this Lesser General Public License (also called "this License").+Each licensee is addressed as "you".++ A "library" means a collection of software functions and/or data+prepared so as to be conveniently linked with application programs+(which use some of those functions and data) to form executables.++ The "Library", below, refers to any such software library or work+which has been distributed under these terms. A "work based on the+Library" means either the Library or any derivative work under+copyright law: that is to say, a work containing the Library or a+portion of it, either verbatim or with modifications and/or translated+straightforwardly into another language. (Hereinafter, translation is+included without limitation in the term "modification".)++ "Source code" for a work means the preferred form of the work for+making modifications to it. For a library, complete source code means+all the source code for all modules it contains, plus any associated+interface definition files, plus the scripts used to control compilation+and installation of the library.++ Activities other than copying, distribution and modification are not+covered by this License; they are outside its scope. The act of+running a program using the Library is not restricted, and output from+such a program is covered only if its contents constitute a work based+on the Library (independent of the use of the Library in a tool for+writing it). Whether that is true depends on what the Library does+and what the program that uses the Library does.++ 1. You may copy and distribute verbatim copies of the Library's+complete source code as you receive it, in any medium, provided that+you conspicuously and appropriately publish on each copy an+appropriate copyright notice and disclaimer of warranty; keep intact+all the notices that refer to this License and to the absence of any+warranty; and distribute a copy of this License along with the+Library.++ You may charge a fee for the physical act of transferring a copy,+and you may at your option offer warranty protection in exchange for a+fee.++ 2. You may modify your copy or copies of the Library or any portion+of it, thus forming a work based on the Library, and copy and+distribute such modifications or work under the terms of Section 1+above, provided that you also meet all of these conditions:++ a) The modified work must itself be a software library.++ b) You must cause the files modified to carry prominent notices+ stating that you changed the files and the date of any change.++ c) You must cause the whole of the work to be licensed at no+ charge to all third parties under the terms of this License.++ d) If a facility in the modified Library refers to a function or a+ table of data to be supplied by an application program that uses+ the facility, other than as an argument passed when the facility+ is invoked, then you must make a good faith effort to ensure that,+ in the event an application does not supply such function or+ table, the facility still operates, and performs whatever part of+ its purpose remains meaningful.++ (For example, a function in a library to compute square roots has+ a purpose that is entirely well-defined independent of the+ application. Therefore, Subsection 2d requires that any+ application-supplied function or table used by this function must+ be optional: if the application does not supply it, the square+ root function must still compute square roots.)++These requirements apply to the modified work as a whole. If+identifiable sections of that work are not derived from the Library,+and can be reasonably considered independent and separate works in+themselves, then this License, and its terms, do not apply to those+sections when you distribute them as separate works. But when you+distribute the same sections as part of a whole which is a work based+on the Library, the distribution of the whole must be on the terms of+this License, whose permissions for other licensees extend to the+entire whole, and thus to each and every part regardless of who wrote+it.++Thus, it is not the intent of this section to claim rights or contest+your rights to work written entirely by you; rather, the intent is to+exercise the right to control the distribution of derivative or+collective works based on the Library.++In addition, mere aggregation of another work not based on the Library+with the Library (or with a work based on the Library) on a volume of+a storage or distribution medium does not bring the other work under+the scope of this License.++ 3. You may opt to apply the terms of the ordinary GNU General Public+License instead of this License to a given copy of the Library. To do+this, you must alter all the notices that refer to this License, so+that they refer to the ordinary GNU General Public License, version 2,+instead of to this License. (If a newer version than version 2 of the+ordinary GNU General Public License has appeared, then you can specify+that version instead if you wish.) Do not make any other change in+these notices.++ Once this change is made in a given copy, it is irreversible for+that copy, so the ordinary GNU General Public License applies to all+subsequent copies and derivative works made from that copy.++ This option is useful when you wish to copy part of the code of+the Library into a program that is not a library.++ 4. You may copy and distribute the Library (or a portion or+derivative of it, under Section 2) in object code or executable form+under the terms of Sections 1 and 2 above provided that you accompany+it with the complete corresponding machine-readable source code, which+must be distributed under the terms of Sections 1 and 2 above on a+medium customarily used for software interchange.++ If distribution of object code is made by offering access to copy+from a designated place, then offering equivalent access to copy the+source code from the same place satisfies the requirement to+distribute the source code, even though third parties are not+compelled to copy the source along with the object code.++ 5. A program that contains no derivative of any portion of the+Library, but is designed to work with the Library by being compiled or+linked with it, is called a "work that uses the Library". Such a+work, in isolation, is not a derivative work of the Library, and+therefore falls outside the scope of this License.++ However, linking a "work that uses the Library" with the Library+creates an executable that is a derivative of the Library (because it+contains portions of the Library), rather than a "work that uses the+library". The executable is therefore covered by this License.+Section 6 states terms for distribution of such executables.++ When a "work that uses the Library" uses material from a header file+that is part of the Library, the object code for the work may be a+derivative work of the Library even though the source code is not.+Whether this is true is especially significant if the work can be+linked without the Library, or if the work is itself a library. The+threshold for this to be true is not precisely defined by law.++ If such an object file uses only numerical parameters, data+structure layouts and accessors, and small macros and small inline+functions (ten lines or less in length), then the use of the object+file is unrestricted, regardless of whether it is legally a derivative+work. (Executables containing this object code plus portions of the+Library will still fall under Section 6.)++ Otherwise, if the work is a derivative of the Library, you may+distribute the object code for the work under the terms of Section 6.+Any executables containing that work also fall under Section 6,+whether or not they are linked directly with the Library itself.++ 6. As an exception to the Sections above, you may also combine or+link a "work that uses the Library" with the Library to produce a+work containing portions of the Library, and distribute that work+under terms of your choice, provided that the terms permit+modification of the work for the customer's own use and reverse+engineering for debugging such modifications.++ You must give prominent notice with each copy of the work that the+Library is used in it and that the Library and its use are covered by+this License. You must supply a copy of this License. If the work+during execution displays copyright notices, you must include the+copyright notice for the Library among them, as well as a reference+directing the user to the copy of this License. Also, you must do one+of these things:++ a) Accompany the work with the complete corresponding+ machine-readable source code for the Library including whatever+ changes were used in the work (which must be distributed under+ Sections 1 and 2 above); and, if the work is an executable linked+ with the Library, with the complete machine-readable "work that+ uses the Library", as object code and/or source code, so that the+ user can modify the Library and then relink to produce a modified+ executable containing the modified Library. (It is understood+ that the user who changes the contents of definitions files in the+ Library will not necessarily be able to recompile the application+ to use the modified definitions.)++ b) Use a suitable shared library mechanism for linking with the+ Library. A suitable mechanism is one that (1) uses at run time a+ copy of the library already present on the user's computer system,+ rather than copying library functions into the executable, and (2)+ will operate properly with a modified version of the library, if+ the user installs one, as long as the modified version is+ interface-compatible with the version that the work was made with.++ c) Accompany the work with a written offer, valid for at+ least three years, to give the same user the materials+ specified in Subsection 6a, above, for a charge no more+ than the cost of performing this distribution.++ d) If distribution of the work is made by offering access to copy+ from a designated place, offer equivalent access to copy the above+ specified materials from the same place.++ e) Verify that the user has already received a copy of these+ materials or that you have already sent this user a copy.++ For an executable, the required form of the "work that uses the+Library" must include any data and utility programs needed for+reproducing the executable from it. However, as a special exception,+the materials to be distributed need not include anything that is+normally distributed (in either source or binary form) with the major+components (compiler, kernel, and so on) of the operating system on+which the executable runs, unless that component itself accompanies+the executable.++ It may happen that this requirement contradicts the license+restrictions of other proprietary libraries that do not normally+accompany the operating system. Such a contradiction means you cannot+use both them and the Library together in an executable that you+distribute.++ 7. You may place library facilities that are a work based on the+Library side-by-side in a single library together with other library+facilities not covered by this License, and distribute such a combined+library, provided that the separate distribution of the work based on+the Library and of the other library facilities is otherwise+permitted, and provided that you do these two things:++ a) Accompany the combined library with a copy of the same work+ based on the Library, uncombined with any other library+ facilities. This must be distributed under the terms of the+ Sections above.++ b) Give prominent notice with the combined library of the fact+ that part of it is a work based on the Library, and explaining+ where to find the accompanying uncombined form of the same work.++ 8. You may not copy, modify, sublicense, link with, or distribute+the Library except as expressly provided under this License. Any+attempt otherwise to copy, modify, sublicense, link with, or+distribute the Library is void, and will automatically terminate your+rights under this License. However, parties who have received copies,+or rights, from you under this License will not have their licenses+terminated so long as such parties remain in full compliance.++ 9. You are not required to accept this License, since you have not+signed it. However, nothing else grants you permission to modify or+distribute the Library or its derivative works. These actions are+prohibited by law if you do not accept this License. Therefore, by+modifying or distributing the Library (or any work based on the+Library), you indicate your acceptance of this License to do so, and+all its terms and conditions for copying, distributing or modifying+the Library or works based on it.++ 10. Each time you redistribute the Library (or any work based on the+Library), the recipient automatically receives a license from the+original licensor to copy, distribute, link with or modify the Library+subject to these terms and conditions. You may not impose any further+restrictions on the recipients' exercise of the rights granted herein.+You are not responsible for enforcing compliance by third parties with+this License.++ 11. If, as a consequence of a court judgment or allegation of patent+infringement or for any other reason (not limited to patent issues),+conditions are imposed on you (whether by court order, agreement or+otherwise) that contradict the conditions of this License, they do not+excuse you from the conditions of this License. If you cannot+distribute so as to satisfy simultaneously your obligations under this+License and any other pertinent obligations, then as a consequence you+may not distribute the Library at all. For example, if a patent+license would not permit royalty-free redistribution of the Library by+all those who receive copies directly or indirectly through you, then+the only way you could satisfy both it and this License would be to+refrain entirely from distribution of the Library.++If any portion of this section is held invalid or unenforceable under any+particular circumstance, the balance of the section is intended to apply,+and the section as a whole is intended to apply in other circumstances.++It is not the purpose of this section to induce you to infringe any+patents or other property right claims or to contest validity of any+such claims; this section has the sole purpose of protecting the+integrity of the free software distribution system which is+implemented by public license practices. Many people have made+generous contributions to the wide range of software distributed+through that system in reliance on consistent application of that+system; it is up to the author/donor to decide if he or she is willing+to distribute software through any other system and a licensee cannot+impose that choice.++This section is intended to make thoroughly clear what is believed to+be a consequence of the rest of this License.++ 12. If the distribution and/or use of the Library is restricted in+certain countries either by patents or by copyrighted interfaces, the+original copyright holder who places the Library under this License may add+an explicit geographical distribution limitation excluding those countries,+so that distribution is permitted only in or among countries not thus+excluded. In such case, this License incorporates the limitation as if+written in the body of this License.++ 13. The Free Software Foundation may publish revised and/or new+versions of the Lesser General Public License from time to time.+Such new versions will be similar in spirit to the present version,+but may differ in detail to address new problems or concerns.++Each version is given a distinguishing version number. If the Library+specifies a version number of this License which applies to it and+"any later version", you have the option of following the terms and+conditions either of that version or of any later version published by+the Free Software Foundation. If the Library does not specify a+license version number, you may choose any version ever published by+the Free Software Foundation.++ 14. If you wish to incorporate parts of the Library into other free+programs whose distribution conditions are incompatible with these,+write to the author to ask for permission. For software which is+copyrighted by the Free Software Foundation, write to the Free+Software Foundation; we sometimes make exceptions for this. Our+decision will be guided by the two goals of preserving the free status+of all derivatives of our free software and of promoting the sharing+and reuse of software generally.++ NO WARRANTY++ 15. BECAUSE THE LIBRARY IS LICENSED FREE OF CHARGE, THERE IS NO+WARRANTY FOR THE LIBRARY, TO THE EXTENT PERMITTED BY APPLICABLE LAW.+EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR+OTHER PARTIES PROVIDE THE LIBRARY "AS IS" WITHOUT WARRANTY OF ANY+KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR+PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE+LIBRARY IS WITH YOU. SHOULD THE LIBRARY PROVE DEFECTIVE, YOU ASSUME+THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION.++ 16. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN+WRITING WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MAY MODIFY+AND/OR REDISTRIBUTE THE LIBRARY AS PERMITTED ABOVE, BE LIABLE TO YOU+FOR DAMAGES, INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR+CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OR INABILITY TO USE THE+LIBRARY (INCLUDING BUT NOT LIMITED TO LOSS OF DATA OR DATA BEING+RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD PARTIES OR A+FAILURE OF THE LIBRARY TO OPERATE WITH ANY OTHER SOFTWARE), EVEN IF+SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH+DAMAGES.++ END OF TERMS AND CONDITIONS++ How to Apply These Terms to Your New Libraries++ If you develop a new library, and you want it to be of the greatest+possible use to the public, we recommend making it free software that+everyone can redistribute and change. You can do so by permitting+redistribution under these terms (or, alternatively, under the terms of the+ordinary General Public License).++ To apply these terms, attach the following notices to the library. It is+safest to attach them to the start of each source file to most effectively+convey the exclusion of warranty; and each file should have at least the+"copyright" line and a pointer to where the full notice is found.++ <one line to give the library's name and a brief idea of what it does.>+ Copyright (C) <year> <name of author>++ This library is free software; you can redistribute it and/or+ modify it under the terms of the GNU Lesser General Public+ License as published by the Free Software Foundation; either+ version 2.1 of the License, or (at your option) any later version.++ This library is distributed in the hope that it will be useful,+ but WITHOUT ANY WARRANTY; without even the implied warranty of+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU+ Lesser General Public License for more details.++ You should have received a copy of the GNU Lesser General Public+ License along with this library; if not, write to the Free Software+ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA++Also add information on how to contact you by electronic and paper mail.++You should also get your employer (if you work as a programmer) or your+school, if any, to sign a "copyright disclaimer" for the library, if+necessary. Here is a sample; alter the names:++ Yoyodyne, Inc., hereby disclaims all copyright interest in the+ library `Frob' (a library for tweaking knobs) written by James Random Hacker.++ <signature of Ty Coon>, 1 April 1990+ Ty Coon, President of Vice++That's all there is to it!
+ COPYRIGHT view
@@ -0,0 +1,18 @@+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++ This library is free software; you can redistribute it and/or+ modify it under the terms of the GNU Lesser General Public+ License as published by the Free Software Foundation; either+ version 2.1 of the License, or (at your option) any later version.++ This library is distributed in the hope that it will be useful,+ but WITHOUT ANY WARRANTY; without even the implied warranty of+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU+ Lesser General Public License for more details.++ You should have received a copy of the GNU Lesser General Public+ License along with this library; if not, write to the Free Software+ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA++Please see COPYING for the license.+
+ Control/Concurrent/ConcurrentUISupport.hs view
@@ -0,0 +1,227 @@+{- +Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru> + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +-- | A bottleneck for UI from multiple threads. Prevents mixing of output +-- messages. Also makes possible a primitive form of input. By mastering +-- @'cuisUIInput_E'@ and @'cuisUIOutput_E'@ it's possible +-- to construct @'ConcurrentUISupport'@ with different than UI I/O. +module Control.Concurrent.ConcurrentUISupport where + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.STM +import Control.Monad +import qualified Control.Exception as E +import Data.List +import qualified Data.Map as M +import Data.Map (Map) +import Data.MyHelpers +import Data.Typeable +import System.IO + +-- * ConcurrentUISupport administration + +type AskerID = String +type ReadFromConsoleStr = String +type ThreadFinisher = IO () + +data CUISCommand = + NewPage_CUISC + | PlusLine_CUISC + | PutStrLn_CUISC String + | ReadStr_CUISC AskerID deriving (Eq) + +data ConcurrentUISupport = + ConcurrentUISupport { + -- | Supported threads are welcome to write their commands here. + -- It is recommended not to use this channel directly - + -- use @'cuisWrite'@, @'cuisReadLn'@ and other mediators instead. + cuisInputTChan :: TChan [CUISCommand] + -- | Map of feedback chans leading to threads, that issue the @'ReadStr_CUISC'@ commands. + , cuisOutputChans :: TVar (Map AskerID (TChan ReadFromConsoleStr)) + -- | Call this, when you want to finish supporter thread. + , cuisFinish :: ThreadFinisher + -- | Was @'cuisFinish'@ called? + , cuisSupporterStopCheck :: STM Bool + -- | What to do, when supporter thread finishes? + , cuisFinalizer :: IO () + -- | How to input, when @'ReadStr_CUISC'@ command comes? + , cuisUIInput_E :: Either (STM String) (IO String) + -- | How to output, when @'PutStrLn_CUISC'@ command comes? + , cuisUIOutput_E :: Either (String -> STM ()) (String -> IO ()) + -- | If set, then call it on each @'NewPage_CUISC'@ command. + , cuisPressAnyKey_uiCmd :: Maybe (IO ()) + -- | For @(Just i)@, if count of printed lines is > @i@, + -- then automatically issue @'NewPage_CUISC'@ command and + -- set lines counter to 0. + , cuisPressAnyKey_linesCountCond :: Maybe Int + } + +defaultConcurrentUISupportIO :: IO ConcurrentUISupport +defaultConcurrentUISupportIO = do + input_tch <- atomically newTChan + breaker_tv <- atomically $ newTVar False + out_chans <- atomically $ newTVar (M.empty) + return ConcurrentUISupport { + cuisInputTChan = input_tch + , cuisOutputChans = out_chans + , cuisFinish = atomically $ writeTVar breaker_tv True + , cuisFinalizer = return () + , cuisSupporterStopCheck = readTVar breaker_tv + , cuisUIInput_E = Right getLine + , cuisUIOutput_E = Right putStrLn + , cuisPressAnyKey_uiCmd = Just pressAnyKey + , cuisPressAnyKey_linesCountCond = Just 22 + } + +-- | UI in console + log to file. +defaultConcurrentUISupportIO_logInFile :: FilePath -> IOMode -> IO ConcurrentUISupport +defaultConcurrentUISupportIO_logInFile f_name io_mode = do + input_tch <- atomically newTChan + breaker_tv <- atomically $ newTVar False + out_chans <- atomically $ newTVar (M.empty) + h <- openFile f_name io_mode + hSetBuffering h LineBuffering + return ConcurrentUISupport { + cuisInputTChan = input_tch + , cuisOutputChans = out_chans + , cuisFinish = atomically $ writeTVar breaker_tv True + , cuisFinalizer = hClose h + , cuisSupporterStopCheck = readTVar breaker_tv + , cuisUIInput_E = Right (do { s <- getLine; hPutStrLn h s; return s}) + , cuisUIOutput_E = Right (\ s -> (hPutStrLn h s >> putStrLn s)) + , cuisPressAnyKey_uiCmd = Just pressAnyKey + , cuisPressAnyKey_linesCountCond = Just 22 + } + +-- | Whatever @'cuisUIInput_E'@ is - @Left@ or @Right@ - flattern it into IO. +cuisUIInput :: ConcurrentUISupport -> IO String +cuisUIInput cuis = + case cuisUIInput_E cuis of + Left stm_a -> atomically stm_a + Right io_a -> io_a + +-- | Whatever @'cuisUIOutput_E'@ is - @Left@ or @Right@ - flattern it into IO. +cuisUIOutput :: ConcurrentUISupport -> String -> IO () +cuisUIOutput cuis s = + case cuisUIOutput_E cuis of + Left stm_a -> atomically $ stm_a s + Right io_a -> io_a s + +-- | Not to be confised with @'cuisUIInput'@, which interfaces with user. +-- This /input/ interfaces with threads, that makes use of it. +-- +-- A wrapper around @'cuisInputTChan'@. +cuisInput :: ConcurrentUISupport -> [CUISCommand] -> IO () +cuisInput cuis cmds = atomically $ writeTChan (cuisInputTChan cuis) cmds + +acquirePutStr_fromCmdsList :: [CUISCommand] -> Maybe CUISCommand +acquirePutStr_fromCmdsList cmds = + let find_p e = + case e of + PutStrLn_CUISC _ -> True + _ -> False + in find find_p cmds +acquireReadStr_fromCmdsList :: [CUISCommand] -> Maybe CUISCommand +acquireReadStr_fromCmdsList cmds = + let find_p e = + case e of + ReadStr_CUISC _ -> True + _ -> False + in find find_p cmds + +runConcurrentUISupport :: ConcurrentUISupport -> IO () +runConcurrentUISupport cuis = do + _ <- forkIO $ + let _cycle :: Int -> IO () + _cycle i = do + (stop, mb_cmds) <- atomically $ do + stop <- cuisSupporterStopCheck cuis + case stop of + True -> return (stop, Nothing) + False -> do + cmds <- readTChan $ cuisInputTChan cuis + return (stop, Just cmds) + new_i <- case mb_cmds of + Nothing -> return i + Just cmds -> do + -- Process PutStrLn_CUISC + case acquirePutStr_fromCmdsList cmds of + Nothing -> return () + Just (PutStrLn_CUISC s) -> cuisUIOutput cuis s + -- Process ReadStr_CUISC + case acquireReadStr_fromCmdsList cmds of + Nothing -> return () + Just (ReadStr_CUISC asker_id) -> do + t <- cuisUIInput cuis + atomically $ do + outs_map <- readTVar (cuisOutputChans cuis) + asker_chan <- case M.lookup asker_id outs_map of + Nothing -> do + asker_chan <- newTChan + modifyTVar_ (cuisOutputChans cuis) (M.insert asker_id asker_chan) + return asker_chan + Just asker_chan -> return asker_chan + writeTChan asker_chan t + -- Process NewPage_CUISC, PlusLine_CUISC + let cond_1 = (cuisPressAnyKey_linesCountCond cuis >>= \ bnd -> return (i >= bnd) ) == Just True + let cond_2 = NewPage_CUISC `elem` cmds + case cond_1 || cond_2 of + True -> return 0 << whenJust (cuisPressAnyKey_uiCmd cuis) + False -> return (i + (bool2num $ PlusLine_CUISC `elem` cmds)) + case stop of + True -> return () -- exit + False -> _cycle new_i + in E.finally (_cycle 0) (cuisFinalizer cuis) + ------------------- + return () + +-- * Use these functions to deal with ConcurrentUISupport + +cuisWriteSTM :: ConcurrentUISupport -> String -> STM () +cuisWriteSTM cuis s = + mapM_ (\ _s -> writeTChan (cuisInputTChan cuis) [PlusLine_CUISC, PutStrLn_CUISC _s]) (lines s) + +cuisWrite :: ConcurrentUISupport -> String -> IO () +cuisWrite cuis s = + mapM_ (\ _s -> cuisInput cuis [PlusLine_CUISC, PutStrLn_CUISC _s]) (lines s) + +cuisNewPage :: ConcurrentUISupport -> IO () +cuisNewPage cuis = + cuisInput cuis [NewPage_CUISC] + +cuisReadLn :: ConcurrentUISupport -> AskerID -> IO ReadFromConsoleStr +cuisReadLn cuis asker_id = do + cuisInput cuis [ReadStr_CUISC asker_id, PlusLine_CUISC] + atomically $ do + outs_map <- readTVar $ cuisOutputChans cuis + case M.lookup asker_id outs_map of + Nothing -> retry + Just asker_chan -> readTChan asker_chan + +------------------------------------------------------------ +-- * Few helpers for dealing with exceptions + +reportExceptionIfAny :: (String -> IO ()) -> String -> IO a -> IO a +reportExceptionIfAny reportStr caller_f_name = E.handle (\ se@(E.SomeException e) -> reportStr ("An error occurred in function '" ++ caller_f_name ++ "'. Type: " ++ (show $ typeRepTyCon $ typeOf e) ++ ". Representation: " ++ show se) >> E.throw (se :: E.SomeException)) + +printExceptionIfAny :: String -> IO a -> IO a +printExceptionIfAny = reportExceptionIfAny putStrLn + +reportExceptionIfAny_2 :: ConcurrentUISupport -> String -> IO a -> IO a +reportExceptionIfAny_2 cuis = reportExceptionIfAny (cuisWrite cuis) + +reportExceptionIfAnySTM :: (String -> STM ()) -> String -> STM a -> STM a +reportExceptionIfAnySTM reportStr caller_f_name stma = catchSTM stma (\ se@(E.SomeException e) -> reportStr ("An error occurred in function '" ++ caller_f_name ++ "'. Type: " ++ (show $ typeRepTyCon $ typeOf e) ++ ". Representation: " ++ show se) >> E.throw (se :: E.SomeException)) + +reportExceptionIfAnySTM_2 :: ConcurrentUISupport -> String -> STM a -> STM a +reportExceptionIfAnySTM_2 cuis = reportExceptionIfAnySTM (cuisWriteSTM cuis) +
+ Control/Concurrent/PriorityChansConverger.hs view
@@ -0,0 +1,20 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++module Control.Concurrent.PriorityChansConverger (+ module Control.Concurrent.PriorityChansConverger.PriorityChansConverger+ , module Control.Concurrent.PriorityChansConverger.Commons+ , module Control.Concurrent.STM.TChan.ExtsCommons+ ) where++import Control.Concurrent.PriorityChansConverger.Commons+import Control.Concurrent.PriorityChansConverger.PriorityChansConverger+import Control.Concurrent.STM.TChan.ExtsCommons
+ Control/Concurrent/PriorityChansConverger/Commons.hs view
@@ -0,0 +1,26 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++module Control.Concurrent.PriorityChansConverger.Commons where++import Control.Concurrent.STM.TChan.ExtsCommons++type StartPriority = ChanContentAmountMeasure+type CurrentPriority = StartPriority+type PCC_ContainsGivenSpecification = Bool+type NothingElseIsInStruct = Bool+type InterruptShouldWe = Bool++data FailureReasonWPCC = BadKey_FRWPCC | ChanFull_FRWPCC+instance Show FailureReasonWPCC where+ show fr = case fr of+ BadKey_FRWPCC -> "Error in adressing priority channel in converger: bad key."+ ChanFull_FRWPCC -> "There's no free space in priority channel - it's full."
+ Control/Concurrent/PriorityChansConverger/PriorityChansConverger.hs view
@@ -0,0 +1,284 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++-- | This module wraps STM PCC with IO.+-- Some locks are added to balance heavyweight transactions, to reduce \"waste+-- efforts\".+module Control.Concurrent.PriorityChansConverger.PriorityChansConverger where++import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.PriorityChansConverger.Commons+import Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM+import Control.Concurrent.ConcurrentUISupport+import Control.Concurrent.STM+import Control.Concurrent.STM.TChan.ExtsCommons+import Control.Exception+import Control.Monad+import Data.IORef+import Data.List+import qualified Data.Map as M+import Data.Map (Map)+import Data.MyHelpers+import System.IO++__locksObserverEnabled = False+----------------------------------------++-- * PriorityChansConverger ADT and it's administration++data PCC_Lock = PCC_Lock | PCC_IOInterruptableRead | PCC_IOInterruptableWrite deriving (Eq, Ord, Show)++-- | Wrapper around @'PriorityChansConvergerSTM'@. Added locks. @'Show'@ instance+-- is defined using @'unsafePerformIO'@.+data Ord k => PriorityChansConverger k e = PriorityChansConverger {+ pccSTM :: PriorityChansConvergerSTM k e+ , pccLocks :: Map PCC_Lock (MVar ())+}++reportIfPossible :: Ord k => PriorityChansConverger k e -> String -> IO ()+reportIfPossible pcc s =+ case pccConcurrentUISupport $ pccSTM pcc of+ Nothing -> return ()+ Just cuis -> cuisWrite cuis s++maybeObserveExceptions :: Ord k => PriorityChansConverger k e -> FunctionName -> IO a -> IO a+maybeObserveExceptions pcc f_name io_a =+ case pccConcurrentUISupport $ pccSTM pcc of+ Nothing -> io_a+ Just cuis -> reportExceptionIfAny_2 cuis f_name io_a++-- | Wrapper around @'newPriorityChansConverger_wCUIS'@ constructor.+--+-- @newPriorityChansConverger = 'newPriorityChansConverger_wCUIS' Nothing@+newPriorityChansConverger :: Ord k => Map k (StartPriority, ChanCapacity) -> IO (PriorityChansConverger k e)+newPriorityChansConverger = newPriorityChansConverger_wCUIS Nothing++-- | Wrapper around @'newPriorityChansConvergerSTM_wCUIS'@ constructor.+-- With an option to enable utility @'ConcurrentUISupport'@+-- (which is useful for debugging)+newPriorityChansConverger_wCUIS :: Ord k => Maybe ConcurrentUISupport -> Map k (StartPriority, ChanCapacity) -> IO (PriorityChansConverger k e)+newPriorityChansConverger_wCUIS mb_cuis chan_priors_map = do+ pcc <- let constr = atomically $ newPriorityChansConvergerSTM_wCUIS mb_cuis chan_priors_map+ in case mb_cuis of+ Nothing -> constr+ Just cuis -> reportExceptionIfAny_2 cuis f_name constr+ l <- newMVar ()+ return PriorityChansConverger {+ pccSTM = pcc+ , pccLocks = M.fromList [(PCC_Lock, l)]+ }+ where f_name = "newPriorityChansConverger_wCUIS"++type FunctionName = String+type LockUserName = FunctionName+-- | Sort the locks and lock them.+--+-- WARNING !!! Asynchronous exceptions may cause loss of locks!+lockPCC :: Ord k => LockUserName -> [PCC_Lock] -> PriorityChansConverger k e -> IO ()+lockPCC locks_user_f_name locks pcc =+ mapM_+ (\ l ->+ case M.lookup l (pccLocks pcc) of+ Nothing -> error ("An error occurred in 'lockPCC': no lock for key '" ++ show l ++ "' exists.")+ Just l_mv -> takeMVar l_mv >> when __locksObserverEnabled (reportIfPossible pcc ("Lock '" ++ show l ++ "' taken by function '" ++ locks_user_f_name ++ "'."))+ )+ (map head $ group $ sort locks)++-- | Reverse sort the locks and unlock them.+--+-- WARNING !!! Asynchronous exceptions may cause loss of locks!+unlockPCC :: Ord k => LockUserName -> [PCC_Lock] -> PriorityChansConverger k e -> IO ()+unlockPCC locks_user_f_name locks pcc =+ mapM_+ (\ l ->+ case M.lookup l (pccLocks pcc) of+ Nothing -> error ("An error occurred in 'unlockPCC': no lock for key '" ++ show l ++ "' exists.")+ Just l_mv -> do+ s <- tryPutMVar l_mv ()+ when __locksObserverEnabled $ reportIfPossible pcc $ case s of+ True -> "Lock '" ++ show l ++ "' released by function '" ++ locks_user_f_name ++ "'."+ False -> "Function '" ++ locks_user_f_name ++ "' wanted to release the lock '" ++ show l ++ "', but it appered to be already released."+ )+ (map head $ group $ reverse $ sort locks)++-- | Locking IO action wrapper.+--+-- WARNING !!! Asynchronous exceptions may cause loss of locks!+withLockedDo :: Ord k => LockUserName -> [PCC_Lock] -> PriorityChansConverger k e -> IO a -> IO a+withLockedDo locks_user_f_name locks pcc act = block $ do+ lockPCC locks_user_f_name locks pcc+ a <- finally (unblock act) (unlockPCC locks_user_f_name locks pcc)+ return a++---------------------------------------------------------------------------------------------------------------------------+-- * PriorityChansConverger mutation+---------------------------------------------------------------------------------------------------------------------------++-- | Wrapper around @'isOfStructPCC_STM'@. Used locks: none.+isOfStructPCC :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConverger k e -> IO (PCC_ContainsGivenSpecification, NothingElseIsInStruct)+isOfStructPCC mp pcc = maybeObserveExceptions pcc f_name $ atomically $ isOfStructPCC_STM mp (pccSTM pcc)+ where f_name = "isOfStructPCC"++-- | Wrapper around @'mutatePCC_STM'@. Used locks: @['PCC_Lock']@+mutatePCC :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConverger k e -> IO (PriorityChansConverger k e)+mutatePCC chan_priors_map pcc = withLockedDo f_name [PCC_Lock] pcc $ do+ new_pcc_stm <- maybeObserveExceptions pcc f_name $ atomically $ mutatePCC_STM chan_priors_map (pccSTM pcc)+ return pcc { pccSTM = new_pcc_stm }+ where f_name = "mutatePCC"++---------------------------------------------------------------------------------------------------------------------------+-- * PriorityChansConverger I/O+---------------------------------------------------------------------------------------------------------------------------++-- | Wrapper around @'writeInPCC_STM'@. Used locks: if blocking,+-- then @[]@ else @['PCC_Lock']@.+--+-- If blocking is enabled, may block, when there is no free space+-- in @'PriorityChan'@.+writeInPCC :: Ord k => PermitToBlock -> k -> e -> PriorityChansConverger k e -> IO (Maybe FailureReasonWPCC)+writeInPCC block_dowe k e pcc = maybe_lock $ maybeObserveExceptions pcc f_name $ atomically $ writeInPCC_STM block_dowe k e (pccSTM pcc)+ where maybe_lock = case block_dowe of { True -> id; False -> withLockedDo f_name [PCC_Lock] pcc }+ f_name = "writeInPCC"++-- | Wrapper around @'interruptableWriteInPCC_STM'@. Used locks: @[]@+--+-- Blocking write.+--+-- Do not put the control of the @'STM' 'InterruptShouldWe'@ under @'PCC_Lock'@.+-- Keep there @False@, if you don't want to get @Left ()@ in return.+-- Put there @True@, whenever you want to stop waiting.+stmInterruptableWriteInPCC :: Ord k => STM InterruptShouldWe -> k -> e -> PriorityChansConverger k e -> IO (Either () (Maybe FailureReasonWPCC))+stmInterruptableWriteInPCC stm_interrupter k e pcc = maybeObserveExceptions pcc f_name $ atomically $ interruptableWriteInPCC_STM stm_interrupter k e (pccSTM pcc)+ where f_name = "stmInterruptableWriteInPCC"++-- | Wrapper around @'stmInterruptableWriteInPCC'@. Used locks: @['PCC_IOInterruptableWrite']@+--+-- Blocking read.+--+-- Creates a @'TVar' Bool@ to control the interruptability of write operation.+-- Spawns additional thread, which cyclicly reads given @Chan@ and checks,+-- if it signals terminating condition - if so, then using "interrupter" @TVar@+-- stops trying to write in PCC. Temporary thread gets terminated at the end.+interruptableWriteInPCC :: Ord k => (Chan a, a -> InterruptShouldWe) -> k -> e -> PriorityChansConverger k e -> IO (Either () (Maybe FailureReasonWPCC))+interruptableWriteInPCC (interrupter_chan, interrupter_cond) k e pcc = withLockedDo f_name [PCC_IOInterruptableWrite] pcc $ maybeObserveExceptions pcc f_name $ do+ interrupter_tv <- atomically $ newTVar False+ tmp_thread_interrupter <- newIORef False+ _ <- forkIO $ let _cycle = do+ stop <- readIORef tmp_thread_interrupter+ case stop of+ True -> return ()+ False -> do+ _interrupt <- interrupter_cond `liftM` readChan interrupter_chan+ case _interrupt of+ False -> _cycle+ True -> atomically $ writeTVar interrupter_tv True+ in _cycle+ r <- stmInterruptableWriteInPCC (readTVar interrupter_tv) k e pcc+ writeIORef tmp_thread_interrupter True+ return r+ where f_name = "interruptableWriteInPCC"++-- | Wrapper around @'readFromPCC_STM'@. Used locks: if blocking, then @[]@ else @['PCC_Lock']@.+readFromPCC :: Ord k => PermitToBlock -> PriorityChansConverger k e -> IO (Maybe (k, e))+readFromPCC block_dowe pcc = maybe_lock $ maybeObserveExceptions pcc f_name $ atomically $ readFromPCC_STM block_dowe (pccSTM pcc)+ where maybe_lock = case block_dowe of { True -> id; False -> withLockedDo f_name [PCC_Lock] pcc }+ f_name = "readFromPCC"++-- | Wrapper around @'interruptableReadFromPCC_STM'@. Used locks: @[]@+--+-- Blocking read.+stmInterruptableReadFromPCC :: Ord k => STM InterruptShouldWe -> PriorityChansConverger k e -> IO (Either () (k, e))+stmInterruptableReadFromPCC stm_interrupter pcc = maybeObserveExceptions pcc f_name $ atomically $ interruptableReadFromPCC_STM stm_interrupter (pccSTM pcc)+ where f_name = "stmInterruptableReadFromPCC"++-- | Wrapper around @'stmInterruptableReadFromPCC'@. Used locks: @['PCC_IOInterruptableRead']@+--+-- Blocking read.+--+-- Creates a @'TVar' Bool@ to control the interruptability of read operation.+-- Spawns additional thread, which cyclicly reads given @Chan@ and checks,+-- if it signals terminating condition - if so, then using "interrupter" @TVar@+-- stops trying to write in PCC. Temporary thread gets terminated at the end.+interruptableReadFromPCC :: Ord k => (Chan a, a -> InterruptShouldWe) -> PriorityChansConverger k e -> IO (Either () (k, e))+interruptableReadFromPCC (interrupter_chan, interrupter_cond) pcc = withLockedDo f_name [PCC_IOInterruptableRead] pcc $ maybeObserveExceptions pcc f_name $ do+ interrupter_tv <- atomically $ newTVar False+ tmp_thread_interrupter <- newIORef False+ _ <- forkIO $ let _cycle = do+ stop <- readIORef tmp_thread_interrupter+ case stop of+ True -> return ()+ False -> do+ _interrupt <- interrupter_cond `liftM` readChan interrupter_chan+ case _interrupt of+ False -> _cycle+ True -> atomically $ writeTVar interrupter_tv True+ in _cycle+ r <- stmInterruptableReadFromPCC (readTVar interrupter_tv) pcc+ writeIORef tmp_thread_interrupter True+ return r+ where f_name = "interruptableReadFromPCC"++-- | Wrapper around @'flushPCC2List_STM'@. Used locks: @['PCC_Lock']@+--+-- Lazy (doublecheck that).+flushPCC2List :: Ord k => PriorityChansConverger k e -> IO [(k, e)]+flushPCC2List pcc = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ flushPCC2List_STM (pccSTM pcc)+ where f_name = "flushPCC2List"++-- | Wrapper around @flushPCC2List_STM'@+-- (see "Control.Concurrent.PriorityChansConverger.PriorityChansConverger").+-- Used locks: @['PCC_Lock']@+--+-- Strict. Should be a bit faster than @'flushPCC2List'@.+flushPCC2List' :: Ord k => PriorityChansConverger k e -> IO [(k, e)]+flushPCC2List' pcc = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ flushPCC2List_STM' (pccSTM pcc)+ where f_name = "flushPCC2List\'"++-- | Wrapper around @fastFlushPCC2List_STM'@+-- (see "Control.Concurrent.PriorityChansConverger.PriorityChansConverger").+-- Used locks: @['PCC_Lock']@+--+-- Strict. This is a fast flush, it doesn't deal with priorities.+fastFlushPCC2List' :: Ord k => PriorityChansConverger k e -> IO [(k, e)]+fastFlushPCC2List' pcc = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ fastFlushPCC2List_STM' (pccSTM pcc)+ where f_name = "fastFlushPCC2List\'"++-- | Wrapper around @'isEmptyPCC_STM'@. Used locks: none+isEmptyPCC :: Ord k => PriorityChansConverger k e -> IO Bool+isEmptyPCC pcc = maybeObserveExceptions pcc f_name $ atomically $ isEmptyPCC_STM (pccSTM pcc)+ where f_name = "isEmptyPCC"++-- | Wrapper around @'freeSpaceInPCCInput_STM'@. Used locks: none+freeSpaceInPCCInput :: Ord k => PriorityChansConverger k e -> k -> IO (Maybe ChanContentAmountMeasure)+freeSpaceInPCCInput pcc k = maybeObserveExceptions pcc f_name $ atomically $ freeSpaceInPCCInput_STM (pccSTM pcc) k+ where f_name = "freeSpaceInPCCInput"++-- | Wrapper around @filterOutPCCElements_STM'@+-- (see "Control.Concurrent.PriorityChansConverger.PriorityChansConverger").+-- Used locks: @['PCC_Lock']@+--+-- Strict.+--+-- Take everything, applu filter, return what's left in the PCC,+-- return what's taken out.+filterOutPCCElements' :: Ord k => PriorityChansConverger k e -> (e -> TakeElementOutShouldWe) -> IO [(k, e)]+filterOutPCCElements' pcc p = withLockedDo f_name [PCC_Lock] pcc $ maybeObserveExceptions pcc f_name $ atomically $ filterOutPCCElements_STM' (pccSTM pcc) p+ where f_name = "filterOutPCCElements\'"++---------------------------------------------------------------------------------------------------------------------------+-- * PriorityChansConverger representation+---------------------------------------------------------------------------------------------------------------------------++-- | Wrapper around @'showPCC'@. Used locks: none+showPCC :: (Ord k, Show k, Show e) => PriorityChansConverger k e -> IO String+showPCC pcc = maybeObserveExceptions pcc f_name $ atomically $ showPCC_STM (pccSTM pcc)+ where f_name = "showPCC"
+ Control/Concurrent/PriorityChansConverger/PriorityChansConvergerSTM.hs view
@@ -0,0 +1,428 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++{-# LANGUAGE MultiParamTypeClasses #-}++-- | Abbreviation for the @PriorityChansConverger@ is PCC.+--+-- Based on @STM.TChan@s, extended with capacity control+-- (see @"Control.Concurrent.STM.TChan.TChanB"@).+--+-- When user reads from the PCC, the choice is made - from which channel+-- to read. System selects a nonempty channel, whose /(CurrentPriority,+-- StartPriority)/ tuple is max. The side effect of the channel selection+-- is it's /CurrentPriority/ decrease by one, if it's value becomes+-- less than one, then the /CurrentPriority/ is set to /StartPriority/.+module Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM where++import Control.Concurrent.ConcurrentUISupport+import Control.Concurrent.PriorityChansConverger.Commons+import Control.Concurrent.STM+import Control.Concurrent.STM.TChan.TChanB+import Control.Monad+import Data.Foldable+import qualified Data.Map as M+import Data.Map (Map)+import Data.Maybe+import Data.MyHelpers++----------------------------------------+-- * PriorityChansConverger ADT and it's administration++-- | Decrease @'CurrentPriority'@ by one, if it's value becomes+-- less than one, then the @'CurrentPriority'@ is set to @'StartPriority'@.+decPriority :: (CurrentPriority, StartPriority) -> (CurrentPriority, StartPriority)+decPriority (cur, start) =+ ( case cur <= 1 of+ True -> start+ False -> cur - 1+ , start+ )++-- Core ADT of this package. It is thought to be wrapped in a IO ADT,+-- though, in order to optimize performance+-- (which is done in the+-- @Control.Concurrent.PriorityChansConverger.PriorityChansConverger@+-- module).+data Ord k => PriorityChansConvergerSTM k e = PriorityChansConvergerSTM {+ -- | All channels. Convenient to use for input.+ -- Map keys == @'pcID'@ of elements.+ pccStruct :: Map k (PriorityChan k e)+ -- | Empty channels. Map keys == @'pcPriorityTV'@ of elements.+ , pccEmpties :: TVar (Map (CurrentPriority, StartPriority) [PriorityChan k e])+ -- | Nonempty channels. Map keys == @'pcPriorityTV'@ of elements.+ -- Convenient for taking output.+ , pccNonEmpties :: TVar (Map (CurrentPriority, StartPriority) [PriorityChan k e])+ -- | For testing and debugging purposes.+ , pccConcurrentUISupport :: Maybe ConcurrentUISupport+}++-- | PriorityChan = ID + @'TChanB'@ + @'TVar' ('CurrentPriority', 'StartPriority')@+data PriorityChan k e = PriorityChan {+ pcID :: k+ , pcChanB :: TChanB e+ , pcPriorityTV :: TVar (CurrentPriority, StartPriority)+}++-- | @'pcLoad' = 'chanLoad' . 'pcChanB'@+pcLoad :: PriorityChan k e -> STM ChanLoad+pcLoad = chanLoad . pcChanB++-- | @'pcCapacity' = 'chanCapacity' . 'pcChanB'@+pcCapacity :: PriorityChan k e -> STM ChanCapacity+pcCapacity = chanCapacity . pcChanB++-- | @'pcIsEmpty' = 'isEmptyTChanB' . 'pcChanB'@+pcIsEmpty :: PriorityChan k e -> STM Bool+pcIsEmpty = isEmptyTChanB . pcChanB++-- | @'pcPriority' = 'readTVar' . 'pcPriorityTV'@+pcPriority :: PriorityChan k e -> STM (CurrentPriority, StartPriority)+pcPriority = readTVar . pcPriorityTV++-- | @'pcStartPriority' pc = snd `liftM` 'pcPriority' pc@+pcStartPriority :: PriorityChan k e -> STM StartPriority+pcStartPriority pc = snd `liftM` pcPriority pc++-- | Constructor.+newPriorityChan :: k -> StartPriority -> ChanCapacity -> STM (PriorityChan k e)+newPriorityChan _id _pcStartPrior _pcCapac = do+ _pcChanB <- newTChanB _pcCapac+ _pcPriorityTV <- newTVar (_pcStartPrior, _pcStartPrior)+ return PriorityChan {+ pcID = _id+ , pcChanB = _pcChanB+ , pcPriorityTV = _pcPriorityTV+ }++-- | PCC constructor with an option to enable utility @'ConcurrentUISupport'@+-- (which is useful for debugging).+newPriorityChansConvergerSTM_wCUIS :: Ord k => Maybe ConcurrentUISupport -> Map k (StartPriority, ChanCapacity) -> STM (PriorityChansConvergerSTM k e)+newPriorityChansConvergerSTM_wCUIS mb_cuis chan_priors_map = do+ (_pccStruct, _pccEmpties) <-+ foldrM (\ (k, (p, c)) (struct, state) -> do+ pc <- newPriorityChan k p c+ let new_struct = M.insert k pc struct+ let new_state = M.unionWith (++) state (M.singleton (p, p) [pc])+ return (new_struct, new_state)+ )+ (M.empty, M.empty)+ (M.toList chan_priors_map)+ _pccEmptiesTV <- newTVar _pccEmpties+ _pccNonemptiesTV <- newTVar M.empty+ return PriorityChansConvergerSTM {+ pccStruct = _pccStruct+ , pccEmpties = _pccEmptiesTV+ , pccNonEmpties = _pccNonemptiesTV+ , pccConcurrentUISupport = mb_cuis+ }++-- | PCC constructor. @'ConcurrentUISupport'@ is off.+--+-- @newPriorityChansConvergerSTM = 'newPriorityChansConvergerSTM_wCUIS' Nothing@+newPriorityChansConvergerSTM :: Ord k => Map k (StartPriority, ChanCapacity) -> STM (PriorityChansConvergerSTM k e)+newPriorityChansConvergerSTM = newPriorityChansConvergerSTM_wCUIS Nothing++---------------------------------------------------------------------------------------------------------------------------+-- * PriorityChansConvergerSTM mutation+---------------------------------------------------------------------------------------------------------------------------++-- | Checks:+--+-- * If PCC contains all channels, constructable by given specification.+--+-- * If PCC contains nothing else, than mentioned in the given specification.+isOfStructPCC_STM :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConvergerSTM k e -> STM (PCC_ContainsGivenSpecification, NothingElseIsInStruct)+isOfStructPCC_STM mp pcc = do+ (sogood_sofar, left_overs) <-+ foldrM (\ (k, (p, c)) (sogood_sofar, mp) ->+ case takeFromMap k mp of+ (Nothing, _) -> return (False, mp)+ (Just pc, rest_mp) -> do+ cap <- pcCapacity pc+ start <- pcStartPriority pc+ return (sogood_sofar && cap == c && start == p, rest_mp)+ )+ (True, pccStruct pcc)+ (M.toList mp)+ return (sogood_sofar, M.null left_overs)++-- | Updates PCC with given specification. New channels may be added+-- or existing alerted.+-- But no operation for removing channels.+mutatePCC_STM :: Ord k => Map k (StartPriority, ChanCapacity) -> PriorityChansConvergerSTM k e -> STM (PriorityChansConvergerSTM k e)+mutatePCC_STM chan_priors_map pcc = do+ new_struct_map <- getNewStruct+ return pcc { pccStruct = new_struct_map }+ where+ getNewStruct = foldrM+ (\ (k, (new_start, cap)) struct_map -> do+ (new_struct_map, pc, mb_old_prior) <-+ case M.lookup k struct_map of+ Nothing -> do+ pc <- newPriorityChan k new_start cap+ return $ (M.insert k pc struct_map, pc, Nothing)+ Just pc -> do+ (_cur, _start) <- readTVar (pcPriorityTV pc)+ let new_prior = (_cur, new_start)+ writeTVar (pcPriorityTV pc) (_cur, new_start)+ changeChanCapacity cap (pcChanB pc)+ return (struct_map, pc, Just (_cur, _start))+ emp <- pcIsEmpty pc+ case mb_old_prior of+ Nothing -> modifyTVar_+ (pccEmpties pcc)+ (M.unionWith (++) (M.singleton (new_start, new_start) [pc]))+ Just old_key@(cur, old_start) -> modifyTVar_+ (case emp of+ True -> pccEmpties pcc+ False -> pccNonEmpties pcc+ )+ (\ mp0 ->+ let mp1 = M.update (\ pc_list ->+ let new_pc_list = filter (\ pc -> pcID pc /= k) pc_list+ in case null new_pc_list of+ True -> Nothing+ False -> Just new_pc_list+ ) old_key mp0+ mp2 = M.unionWith (++) (M.singleton (cur, new_start) [pc]) mp1+ in mp2+ )+ return new_struct_map+ )+ (pccStruct pcc)+ (M.toList chan_priors_map)++---------------------------------------------------------------------------------------------------------------------------+-- * PriorityChansConvergerSTM I/O+---------------------------------------------------------------------------------------------------------------------------++-- | If blocking is on, then blocks, when trying to write to a full channel+-- (where's no free space).+writeInPCC_STM :: Ord k => PermitToBlock -> k -> e -> PriorityChansConvergerSTM k e -> STM (Maybe FailureReasonWPCC)+writeInPCC_STM block_dowe k e pcc =+ case M.lookup k (pccStruct pcc) of+ Nothing -> return $ Just BadKey_FRWPCC+ Just pc -> do+ emp <- pcIsEmpty pc+ wrote <- elasticWriteTChanB block_dowe (pcChanB pc) e+ case wrote of+ False -> return $ Just ChanFull_FRWPCC+ True ->+ case emp of+ False -> return Nothing+ True -> do+ prior_tuple <- pcPriority pc+ modifyTVar_+ (pccEmpties pcc)+ (M.update+ (\ pc_list ->+ case filter (\ pc -> pcID pc /= k) pc_list of+ [] -> Nothing+ new_pc_list -> Just new_pc_list+ )+ prior_tuple+ )+ modifyTVar_+ (pccNonEmpties pcc)+ (M.unionWith (++) (M.singleton prior_tuple [pc]))+ return Nothing++-- | Wrapper around @('writeInPCC_STM' True)@ - blocking write.+-- Uses 'interruptableSTM'. If returns Left, then transaction ended with+-- interrupting condition.+interruptableWriteInPCC_STM :: Ord k => STM InterruptShouldWe -> k -> e -> PriorityChansConvergerSTM k e -> STM (Either () (Maybe FailureReasonWPCC))+interruptableWriteInPCC_STM stm_interrupter k e pcc = do+ mb_r <- interruptableSTM stm_interrupter (writeInPCC_STM True k e pcc)+ return $ case mb_r of+ Nothing -> Left ()+ Just smtng -> Right smtng++-- | If @'PermitToBlock'@ is @True@, then never returns @Nothing@.+readFromPCC_STM :: Ord k => PermitToBlock -> PriorityChansConvergerSTM k e -> STM (Maybe (k, e))+readFromPCC_STM block_dowe pcc = do+ mp <- readTVar $ pccNonEmpties pcc+ case M.maxViewWithKey mp of+ Nothing ->+ case block_dowe of+ True -> retry+ False -> return Nothing+ Just ((prior, pc : rest_pc_list), rest_mp) -> do -- this assumes that on place of "pc : _" can't be []+ mb_e <- fromJust `liftM` tryReadTChanB (pcChanB pc) -- this assumes that if given PriorityChan is in pccNonEmpties map, then it may NOT be empty... orelse, we get an exception here+ emp <- pcIsEmpty pc+ let new_prior = decPriority prior+ case emp of+ True -> modifyTVar_+ (pccEmpties pcc)+ (M.unionWith (++) (M.singleton new_prior [pc]))+ False -> return ()+ let new_mp =+ case emp of+ True ->+ case rest_pc_list of+ [] -> rest_mp+ _ -> M.insert prior rest_pc_list rest_mp+ False ->+ let map_1 =+ case rest_pc_list of+ [] -> rest_mp+ _ -> M.insert prior rest_pc_list rest_mp+ in M.unionWith (++) (M.singleton new_prior [pc]) map_1+ modifyTVar_+ (pccNonEmpties pcc)+ (const new_mp)+ modifyTVar_+ (pcPriorityTV pc)+ (const new_prior)+ return $ Just (pcID pc, mb_e)+ Just ((_, []),_) -> error "An error occurred in 'readFromPCC_STM': empty list in the entry of pccNonEmpties!"+ where f_name = "readFromPCC_STM"++-- | Wrapper around @('readFromPCC_STM' True)@ - blocking read.+-- Uses 'interruptableSTM'. If returns Left, then transaction ended with+-- interrupting condition.+interruptableReadFromPCC_STM :: Ord k => STM InterruptShouldWe -> PriorityChansConvergerSTM k e -> STM (Either () (k, e))+interruptableReadFromPCC_STM stm_interrupter pcc = do+ mb_r <- interruptableSTM stm_interrupter (readFromPCC_STM True pcc)+ return $ case mb_r of+ Nothing -> Left ()+ Just (Just smtng) -> Right smtng+ Just Nothing -> error "An error occurred in 'interruptableReadFromPCC_STM': blocking read returned Nothing!"++-- | Composition of @'readFromPCC_STM'@s. Lazy (doublecheck that).+flushPCC2List_STM :: Ord k => PriorityChansConvergerSTM k e -> STM [(k, e)]+flushPCC2List_STM pcc = reverse `liftM` whileJustM_1 (:) [] (readFromPCC_STM False pcc)++-- | Strict. Should be a bit faster than @'flushPCC2List_STM'@.+flushPCC2List_STM' :: Ord k => PriorityChansConvergerSTM k e -> STM [(k, e)]+flushPCC2List_STM' pcc = do+ ne_pc_mp <- readTVar $ pccNonEmpties pcc+ (e_list, empties) <- readFromNonempties ne_pc_mp+ modifyTVar_ (pccEmpties pcc) (M.unionWith (++) empties)+ return e_list+ where+ readFromNonempties :: Map (CurrentPriority, StartPriority) [PriorityChan k e] -> STM ([(k, e)], Map (CurrentPriority, StartPriority) [PriorityChan k e])+ readFromNonempties mp =+ case M.maxViewWithKey mp of+ Nothing -> return ([], M.empty)+ Just ((prior, pc_list), rest_mp) -> do+ let new_prior = decPriority prior+ (e_list, nonempties, empties) <- foldrM+ (\ pc (elems_accum, nonempties_accum, empties_accum) -> do+ mb_e <- tryReadTChanB (pcChanB pc)+ return $ case mb_e of+ Nothing -> ( elems_accum, nonempties_accum, pc : empties_accum)+ Just e -> ((pcID pc, e) : elems_accum, pc : nonempties_accum, empties_accum)+ )+ ([], [], [])+ pc_list+ let continue_with_map = M.unionWith (++) rest_mp (M.singleton new_prior nonempties)+ let stay_with_map = M.singleton prior empties+ (lower_prior_e_list, rest_of_empties) <- readFromNonempties continue_with_map+ return (e_list ++ lower_prior_e_list, M.unionWith (++) stay_with_map rest_of_empties)++-- | Strict. This is a fast flush, it doesn't deal with priorities.+fastFlushPCC2List_STM' :: Ord k => PriorityChansConvergerSTM k e -> STM [(k, e)]+fastFlushPCC2List_STM' pcc = do+ ne_pc_mp <- readTVar $ pccNonEmpties pcc+ modifyTVar_ (pccEmpties pcc) (M.unionWith (++) ne_pc_mp)+ let pc_lists_list = M.elems ne_pc_mp+ elems_list <- foldlM (foldlM foldlM_r) [] pc_lists_list+ return elems_list+ where+ foldlM_r :: [(k, e)] -> PriorityChan k e -> STM [(k, e)]+ foldlM_r accum pc = do+ let k = pcID pc+ el <- getTChanBContents (pcChanB pc)+ return (accum ++ map (\ e -> (k, e)) el)++isEmptyPCC_STM :: Ord k => PriorityChansConvergerSTM k e -> STM Bool+isEmptyPCC_STM pcc = M.null `liftM` (readTVar $ pccNonEmpties pcc)++-- | Free space in referenced channel = Capacity - Load.+freeSpaceInPCCInput_STM :: Ord k => PriorityChansConvergerSTM k e -> k -> STM (Maybe ChanContentAmountMeasure)+freeSpaceInPCCInput_STM pcc k = do+ case M.lookup k $ pccStruct pcc of+ Nothing -> return Nothing+ Just pc -> do+ cap <- pcCapacity pc+ lo <- pcLoad pc+ return $ Just (cap - lo)++-- | Strict.+--+-- Take everything, applu filter, return what's left in the PCC,+-- return what's taken out.+filterOutPCCElements_STM' :: Ord k => PriorityChansConvergerSTM k e -> (e -> TakeElementOutShouldWe) -> STM [(k, e)]+filterOutPCCElements_STM' pcc p = foldlM _filterOutPCElements [] (M.elems $ pccStruct pcc)+ where+ -- _filterOutPCElements :: [(k, e)] -> PriorityChan k e -> STM [(k, e)]+ _filterOutPCElements takeout_accum pc = do+ takeout_add <- filterOutTChanBElements p (pcChanB pc)+ return $ join [zip (repeat $ pcID pc) takeout_add, takeout_accum]+++---------------------------------------------------------------------------------------------------------------------------+-- * PriorityChansConvergerSTM representation+---------------------------------------------------------------------------------------------------------------------------++-- | Used wrapped in @(unsafePerformIO . atomically)@ in the definition+-- of show instance.+showPCC_STM :: (Ord k, Show k, Show e) => PriorityChansConvergerSTM k e -> STM String+showPCC_STM pcc = do+ let f = (\ pc_map_by_prior -> map (\ (p, pc_list) -> (p, map pcID pc_list)) (M.toList pc_map_by_prior) )+ emps <- f `liftM` (readTVar $ pccEmpties pcc)+ nemps <- f `liftM` (readTVar $ pccNonEmpties pcc)+ let s = "PriorityChansConvergerSTM {"+ ++ "\n | Empties = " ++ (show $ emps)+ ++ "\n | NonEmpties = " ++ (show $ nemps)+ ++ "\n | Chans count = " ++ (show $ M.size $ pccStruct pcc) ++ " chans"+ ++ "\n |+ Channels : "+ s_w_c <- foldrM (\ pc s_accum -> showPC_STM pc >>= \ s_add -> return (s_accum ++ s_add) ) s (M.elems $ pccStruct pcc)+ return (s_w_c ++ "\n}")++-- | Used wrapped in @(unsafePerformIO . atomically)@ in the definition+-- of show instance.+showPC_STM :: (Show e, Show k) => PriorityChan k e -> STM String+showPC_STM pc = do+ (cur, start) <- pcPriority pc+ cap <- pcCapacity pc+ load <- pcLoad pc+ let s = "\n ||+ PriorityChan {"+ ++ "\n ||| ID = " ++ (show $ pcID pc)+ ++ "\n ||| CurrentPriority = " ++ (show cur)+ ++ "\n ||| StartPriority = " ++ (show start)+ ++ "\n ||| Capacity = " ++ (show cap)+ ++ "\n ||| Load = " ++ (show load)+ ++ "\n |||+ Content : "+ elems <- viewChanBContent $ pcChanB pc+ let ch_content_str = (Prelude.concat $ mapWithIndex (\ i e -> "\n |||| -- " ++ show i ++ ") " ++ show e) elems)+ ++ "\n ||- }"+ return (s ++ ch_content_str)++-- instance (Show k, Show e) => Show (PriorityChan k e) where+-- show pc = unsafePerformIO $ printExceptionIfAny "show:PriorityChan" $ atomically $ showPC_STM pc+-- instance (Ord k, Show k, Show e) => Show (PriorityChansConvergerSTM k e) where+-- show pcc = unsafePerformIO $ printExceptionIfAny "show:PriorityChansConvergerSTM" $ atomically $ showPCC_STM pcc++--------------------------------++-- * Helpers++-- | For @interruptableSTM interrupter subj@, when @interrupter@ returns @True@,+-- the transaction returns @Nothing@. Else the result of @subj@ is waited+-- and returned wrapped into @Just@.+interruptableSTM :: STM InterruptShouldWe -> STM a -> STM (Maybe a)+interruptableSTM interrupter subj = do+ ir <- interrupter+ case ir of+ True -> return Nothing+ False -> Just `liftM` subj
+ Control/Concurrent/STM/TChan/ExtsCommons.hs view
@@ -0,0 +1,22 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++module Control.Concurrent.STM.TChan.ExtsCommons where++-- | Bigger than @Int@ might not work, because some list functions are used,+-- that consume @Int@+type ChanContentAmountMeasure = Int+type ChanLoad = ChanContentAmountMeasure+type ChanCapacity = ChanContentAmountMeasure+type PermitToBlock = Bool+type WriteSucceeded = Bool+type TakeElementOutShouldWe = Bool+
+ Control/Concurrent/STM/TChan/TChanB.hs view
@@ -0,0 +1,18 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++module Control.Concurrent.STM.TChan.TChanB (+ module Control.Concurrent.STM.TChan.TChanB_+ , module Control.Concurrent.STM.TChan.ExtsCommons+ ) where++import Control.Concurrent.STM.TChan.TChanB_+import Control.Concurrent.STM.TChan.ExtsCommons
+ Control/Concurrent/STM/TChan/TChanB_.hs view
@@ -0,0 +1,127 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++-- | @'TChanL' = 'TChan' + 'TVar' 'ChanLoad'@+--+-- @'TChanB' = 'TChanL' + 'TVar' 'ChanCapacity'@+module Control.Concurrent.STM.TChan.TChanB_ where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.STM.TChan.TChanL hiding (chanLoad)+import qualified Control.Concurrent.STM.TChan.TChanL as TCHL (chanLoad)+import Data.List+import Data.MyHelpers++--------------------------------------------------------------++data TChanB a =+ TChanB {+ tcbChanL :: TChanL a+ , tcbCapacity :: TVar ChanCapacity+ }++chanCapacity :: TChanB a -> STM ChanCapacity+chanCapacity tchb = readTVar $ tcbCapacity tchb++changeChanCapacity :: ChanCapacity -> TChanB a -> STM ()+changeChanCapacity cap tchb = writeTVar (tcbCapacity tchb) cap++chanLoad :: TChanB a -> STM ChanCapacity+chanLoad tchb = TCHL.chanLoad $ tcbChanL tchb++newTChanB :: ChanCapacity -> STM (TChanB a)+newTChanB cap = do+ tch <- newTChanL+ tv <- newTVar cap+ return TChanB {+ tcbChanL = tch+ , tcbCapacity = tv+ }++newTChanB_IO :: ChanCapacity -> IO (TChanB a)+newTChanB_IO cap = atomically $ newTChanB cap++-- | The \"elastic-\" prefix stands for the opportunity to regulate+-- with option 'PermitToBlock'+elasticWriteTChanB :: PermitToBlock -> TChanB a -> a -> STM WriteSucceeded+elasticWriteTChanB block_dowe tchb a = do+ load <- chanLoad tchb+ cap <- chanCapacity tchb+ case load < cap of+ True -> return True << writeTChanL (tcbChanL tchb) a+ False -> case block_dowe of+ True -> retry+ False -> return False++-- | Wrapper around 'elasticWriteTChanB':+-- @writeTChanB tchb a = return () << elasticWriteTChanB True tchb a@+writeTChanB :: TChanB a -> a -> STM ()+writeTChanB tchb a = return () << elasticWriteTChanB True tchb a++readTChanB :: TChanB a -> STM a+readTChanB tchb = readTChanL (tcbChanL tchb)++isEmptyTChanB :: TChanB a -> STM Bool+isEmptyTChanB tchb = isEmptyTChanL (tcbChanL tchb)++-- | @freeSpace = 'chanCapacity' - 'chanLoad'@+freeSpaceInTChanB :: TChanB a -> STM ChanContentAmountMeasure+freeSpaceInTChanB tchb = do+ load <- chanLoad tchb+ cap <- chanCapacity tchb+ return (cap - load)++tryReadTChanB :: TChanB a -> STM (Maybe a)+tryReadTChanB tchb = tryReadTChanL $ tcbChanL tchb++-- | Wrapper around 'elasticWriteTChanB':+-- @tryWriteTChanB = elasticWriteTChanB False@+tryWriteTChanB :: TChanB a -> a -> STM Bool+tryWriteTChanB = elasticWriteTChanB False++-- | Take everything from channel, and leave it empty.+getTChanBContents :: TChanB a -> STM [a]+getTChanBContents = getTChanLContents . tcbChanL++-- | This procedure is highly uneffective and is dangerous,+-- when @'PermitToBlock'@ is on.+-- Since it won't complete unless there is enough free space in the channnel.+-- Even the part of input that fits won't commit to get into the channel.+-- If capacity isn't enough, it won't ever commit,+-- unless the capacity is made bigger.+elacticWriteList2TChanB :: PermitToBlock -> TChanB a -> [a] -> STM [a]+elacticWriteList2TChanB block_dowe tchb l = do+ available_space <- freeSpaceInTChanB tchb+ let (to_load, rest) = splitAt available_space l+ case block_dowe of+ False -> do+ writeList2TChanL (tcbChanL tchb) to_load+ return rest+ True ->+ case rest of+ [] -> do+ writeList2TChanL (tcbChanL tchb) to_load+ return []+ _ -> retry++-- | Just filter out elements. Is abstracted over implementation,+-- so doesn't suffer from possible inconsistency between capacity+-- and load (when capacity < load).+filterOutTChanBElements :: (a -> TakeElementOutShouldWe) -> TChanB a -> STM [a]+filterOutTChanBElements p tchb = do+ l <- getTChanBContents tchb+ let (takeout, stay) = partition p l+ didnt_fit_back <- writeList2TChanL (tcbChanL tchb) stay+ return takeout++viewChanBContent :: TChanB a -> STM [a]+viewChanBContent tchb = viewChanLContent (tcbChanL tchb)
+ Control/Concurrent/STM/TChan/TChanL.hs view
@@ -0,0 +1,19 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------+++module Control.Concurrent.STM.TChan.TChanL (+ module Control.Concurrent.STM.TChan.TChanL_+ , module Control.Concurrent.STM.TChan.ExtsCommons+ ) where++import Control.Concurrent.STM.TChan.TChanL_+import Control.Concurrent.STM.TChan.ExtsCommons
+ Control/Concurrent/STM/TChan/TChanL_.hs view
@@ -0,0 +1,88 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------+++-- | @'TChanL' = 'TChan' + 'TVar' 'ChanLoad'@+module Control.Concurrent.STM.TChan.TChanL_ where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.STM.TChan.ExtsCommons+import Control.Monad+import Data.Foldable+import Data.List+import Data.MyHelpers++--------------------------------------------------------------++data TChanL a =+ TChanL {+ tclChan :: TChan a+ , tclLoad :: TVar ChanLoad+ }++chanLoad :: TChanL a -> STM ChanLoad+chanLoad tchl = readTVar $ tclLoad tchl++newTChanL :: STM (TChanL a)+newTChanL = do+ tch <- newTChan+ tv <- newTVar 0+ return TChanL {+ tclChan = tch+ , tclLoad = tv+ }++newTChanL_IO :: IO (TChanL a)+newTChanL_IO = atomically newTChanL++writeTChanL :: TChanL a -> a -> STM ()+writeTChanL tchl a = do+ modifyTVar_ (tclLoad tchl) (+ 1)+ writeTChan (tclChan tchl) a++readTChanL :: TChanL a -> STM a+readTChanL tchl = do+ modifyTVar_ (tclLoad tchl) (\ v -> v - 1)+ readTChan (tclChan tchl)++isEmptyTChanL :: TChanL a -> STM Bool+isEmptyTChanL tchl = isEmptyTChan (tclChan tchl)++tryReadTChanL :: TChanL a -> STM (Maybe a)+tryReadTChanL tchl = do+ emp <- isEmptyTChanL tchl+ case emp of+ True -> return Nothing+ False -> Just `liftM` readTChanL tchl++getTChanLContents :: TChanL a -> STM [a]+getTChanLContents tchl = do+ writeTVar (tclLoad tchl) 0+ reverse `liftM` whileJustM_1 (:) [] (tryReadTChan (tclChan tchl))++writeList2TChanL :: TChanL a -> [a] -> STM ()+writeList2TChanL tchl l = do+ load_add <- foldlM (\ load_accum e -> writeTChan (tclChan tchl) e >> return (load_accum + 1)) 0 l+ modifyTVar_ (tclLoad tchl) (\ v -> v + load_add)++filterOutTChanLElements :: (a -> TakeElementOutShouldWe) -> TChanL a -> STM [a]+filterOutTChanLElements p tchl = do+ l <- getTChanLContents tchl+ let (takeout, stay) = partition p l+ writeList2TChanL tchl stay+ return takeout++viewChanLContent :: TChanL a -> STM [a]+viewChanLContent tchl = do+ l <- getTChanLContents tchl+ writeList2TChanL tchl l -- I would gladly use some kind of abortChangesSTM here instead of this+ return l
+ Data/MyHelpers.hs view
@@ -0,0 +1,18 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++module Data.MyHelpers (+ module Data.MyHelpers_+ , module Debug.Trace+ ) where++import Data.MyHelpers_+import Debug.Trace
+ Data/MyHelpers_.hs view
@@ -0,0 +1,384 @@+{-+Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>++All rights reserved.++For license and copyright information, see the file COPYRIGHT+-}++--------------------------------------------------------------------------+--------------------------------------------------------------------------++{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances #-}++module Data.MyHelpers_ where++import Control.Monad+import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Data.Char+import Data.Int+import Data.List+import qualified Data.Map as M+import Data.Map (Map, (!))+import Data.Maybe+import Data.Typeable+import Data.Word++import Foreign.Marshal.Alloc+import Foreign.Ptr+import Foreign.Storable++import System.IO+import System.IO.Unsafe++--------------------------------------++class EQ a b+instance EQ a a++type ErrorMessage = String++infixr 1 <<+(<<) :: Monad m => m b -> m a -> m b+f << x = x >> f++reverse2ple :: (a,b) -> (b,a)+reverse2ple (a,b) = (b,a)++fst3 :: (a,b,c) -> a+snd3 :: (a,b,c) -> b+thrd3 :: (a,b,c) -> c+fst3 (a,_,_) = a+snd3 (_,b,_) = b+thrd3 (_,_,c) = c++apFor2ple :: (a -> a -> a, b -> b -> b) -> (a,b) -> (a,b) -> (a,b)+apFor2ple (af,bf) (a1,b1) (a2,b2) = (af a1 a2, bf b1 b2)++liftTuple :: Monad m => (m a, m b) -> m (a, b)+lift2ple :: Monad m => (m a, m b) -> m (a, b)+lift3ple :: Monad m => (m a, m b, m c) -> m (a, b, c)+lift4ple :: Monad m => (m a, m b, m c, m d) -> m (a, b, c, d)+liftTuple (ma, mb) = do a <- ma+ b <- mb+ return (a,b)+lift2ple = liftTuple+lift3ple (ma, mb, mc) = do a <- ma+ b <- mb+ c <- mc+ return (a,b,c)+lift4ple (ma, mb, mc, md) = do a <- ma+ b <- mb+ c <- mc+ d <- md+ return (a,b,c,d)++ap22ple :: (a, a) -> (a -> b) -> (b, b)+ap22ple (a1,a2) fa = (fa a1, fa a2)+ap22pleM :: Monad m => (a, a) -> (a -> m b) -> m (b, b)+ap22pleM ple f = lift2ple $ ap22ple ple f+ap23ple :: (a, a, a) -> (a -> b) -> (b, b, b)+ap23ple (a1,a2,a3) fa = (fa a1, fa a2, fa a3)+ap23pleM :: Monad m => (a, a, a) -> (a -> m b) -> m (b, b, b)+ap23pleM ple f = lift3ple $ ap23ple ple f+ap24ple :: (a, a, a, a) -> (a -> b) -> (b, b, b, b)+ap24ple (a1,a2,a3,a4) fa = (fa a1, fa a2, fa a3, fa a3)+ap24pleM :: Monad m => (a, a, a, a) -> (a -> m b) -> m (b, b, b, b)+ap24pleM ple f = lift4ple $ ap24ple ple f++apFrom2ple :: ((a -> b), (a -> c)) -> a -> (b, c)+apFrom2ple (f1,f2) a = (f1 a, f2 a)+apFrom3ple :: ((a -> b), (a -> c), (a -> d)) -> a -> (b, c, d)+apFrom3ple (f1,f2,f3) a = (f1 a, f2 a, f3 a)+apFrom4ple :: ((a -> b), (a -> c), (a -> d), (a -> e)) -> a -> (b, c, d, e)+apFrom4ple (f1,f2,f3,f4) a = (f1 a, f2 a, f3 a, f4 a)++liftEither6 :: Either er a -> Either er b -> Either er c -> Either er d -> Either er e -> Either er f -> Either er (a,b,c,d,e,f)+liftEither6 er_or_a er_or_b er_or_c er_or_d er_or_e er_or_f =+ case er_or_a of+ Left er -> Left er+ Right a ->+ case er_or_b of+ Left er -> Left er+ Right b ->+ case er_or_c of+ Left er -> Left er+ Right c ->+ case er_or_d of+ Left er -> Left er+ Right d ->+ case er_or_e of+ Left er -> Left er+ Right e ->+ case er_or_f of+ Left er -> Left er+ Right f -> Right (a,b,c,d,e,f)++liftEither5 :: Either er a -> Either er b -> Either er c -> Either er d -> Either er e -> Either er (a,b,c,d,e)+liftEither5 er_or_a er_or_b er_or_c er_or_d er_or_e =+ case er_or_a of+ Left er -> Left er+ Right a ->+ case er_or_b of+ Left er -> Left er+ Right b ->+ case er_or_c of+ Left er -> Left er+ Right c ->+ case er_or_d of+ Left er -> Left er+ Right d ->+ case er_or_e of+ Left er -> Left er+ Right e -> Right (a,b,c,d,e)++liftEither2 :: Either er a -> Either er b -> Either er (a,b)+liftEither2 er_or_a er_or_b =+ case er_or_a of+ Left er -> Left er+ Right a ->+ case er_or_b of+ Left er -> Left er+ Right b -> Right (a,b)++isLeft :: Either a b -> Bool+isLeft (Left _) = True+isLeft _ = False++isRight :: Either a b -> Bool+isRight = not . isLeft++fromLeft :: Either a b -> a+fromLeft (Left a) = a+fromLeft _ = error "Error occurred, when applied fromLeft to Right!"++fromRight :: Either a b -> b+fromRight (Right b) = b+fromRight _ = error "Error occurred, when applied fromLeft to Right!"++mapWithIndex :: (Int -> a -> b) -> [a] -> [b]+mapWithIndex f l = _mapWithIndex 1 l+ where+ _mapWithIndex _ [] = []+ _mapWithIndex i (h:t) = f i h : _mapWithIndex i t++whileJustM_0 :: Monad m => m (Maybe a) -> m [a]+whileJustM_0 mma = mma >>= \ ma -> case ma of+ Nothing -> return []+ Just a -> (:) a `liftM` whileJustM_0 mma++whileJustM_1 :: Monad m => (a -> b -> b) -> b -> m (Maybe a) -> m b+whileJustM_1 accum_f start_accum src = _whileJustM_1 start_accum+ where+ _whileJustM_1 cur_accum = do+ mb_a <- src+ case mb_a of+ Nothing -> return cur_accum+ Just a -> _whileJustM_1 (accum_f a cur_accum)++takeFromList :: (a -> Bool) -> [a] -> (Maybe a, [a])+takeFromList p l = _takeFromList l+ where+ _takeFromList [] = (Nothing, [])+ _takeFromList (h:t) =+ case p h of+ True -> (Just h, t)+ False -> let (mb_e, processed_t) = _takeFromList t+ in (mb_e, h : processed_t)++takeFromMap :: Ord k => k -> Map k a -> (Maybe a, Map k a)+takeFromMap k m =+ let mb_a = M.lookup k m+ in ( mb_a+ , case isJust mb_a of+ True -> M.delete k m+ False -> m+ )++partition_2 :: Ord k => (a -> Bool) -> (a -> b, a -> c) -> Map k a -> (Map k b, Map k c)+partition_2 p (true_f, false_f) m =+ M.foldWithKey+ (\ k a (true_map, false_map) ->+ case p a of+ True -> (M.insert k (true_f a) true_map, false_map)+ False -> (true_map, M.insert k (false_f a) false_map)+ )+ (M.empty, M.empty)+ m++unionWithKey_2 :: Ord k => (k -> (a, a) -> (a, b)) -> (b -> c -> c) -> c -> Map k a -> Map k a -> (Map k a, c)+unionWithKey_2 unifier side_result_accum_f accum_start m1 m2 =+ foldr (\ (k, a1) (m_accum, side_result_accum) ->+ case takeFromMap k m_accum of+ (Nothing, _) -> (M.insert k a1 m_accum, side_result_accum)+ (Just a2, rest_of_accum_map) ->+ let (ar, side_result_add) = unifier k (a1, a2)+ in (M.insert k ar rest_of_accum_map, side_result_accum_f side_result_add side_result_accum)+ )+ (m2, accum_start)+ (M.toList m1)++unionWithKey_3 :: Ord k => (k -> (Maybe a, Maybe b) -> (c, d)) -> (d -> d -> d) -> d -> Map k a -> Map k b -> (Map k c, d)+unionWithKey_3 unifierF sideResultAccumF accum_start m1 m2 =+ let (m1_result, accum1, m2_cut) = foldr+ (\ (k, a) (m_accum, side_result_accum, m2_cut) ->+ let (mb_b, new_m2_cut) = takeFromMap k m2_cut+ (c, d) = unifierF k (Just a, mb_b)+ new_side_result_accum = sideResultAccumF side_result_accum d+ in (M.insert k c m_accum, new_side_result_accum, new_m2_cut)+ )+ (M.empty, accum_start, m2)+ (M.toList m1)+ (m2_result, accum2) = foldr+ (\ (k, b) (m_accum, side_result_accum) ->+ let (c, d) = unifierF k (Nothing, Just b)+ new_side_result_accum = sideResultAccumF side_result_accum d+ in (M.insert k c m_accum, new_side_result_accum)+ )+ (m1_result, accum1)+ (M.toList m2_cut)+ in (m2_result, accum2)++unionWithKey_4 :: Ord k => (k -> (Maybe a, Maybe b) -> (Maybe c, d)) -> (d -> d -> d) -> d -> Map k a -> Map k b -> (Map k c, d)+unionWithKey_4 unifierF sideResultAccumF accum_start m1 m2 =+ let (m1_result, accum1, m2_cut) = foldr+ (\ (k, a) (m_accum, side_result_accum, m2_cut) ->+ let (mb_b, new_m2_cut) = takeFromMap k m2_cut+ (mb_c, d) = unifierF k (Just a, mb_b)+ new_side_result_accum = sideResultAccumF side_result_accum d+ in ( case mb_c of+ Just c -> M.insert k c m_accum+ Nothing -> m_accum+ , new_side_result_accum, new_m2_cut+ )+ )+ (M.empty, accum_start, m2)+ (M.toList m1)+ (m2_result, accum2) = foldr+ (\ (k, b) (m_accum, side_result_accum) ->+ let (mb_c, d) = unifierF k (Nothing, Just b)+ new_side_result_accum = sideResultAccumF side_result_accum d+ in ( case mb_c of+ Just c -> M.insert k c m_accum+ Nothing -> m_accum+ , new_side_result_accum+ )+ )+ (m1_result, accum1)+ (M.toList m2_cut)+ in (m2_result, accum2)++lookupDeleteFromList :: Eq a => a -> [a] -> (Bool, [a])+lookupDeleteFromList e l =+ let mship = elem e l+ in ( mship+ , case mship of+ True -> delete e l+ False -> l+ )++from2DList :: Ord k => [(k, a)] -> M.Map k [a]+from2DList l = foldl (\ accum_map (k, e) -> M.unionWith (++) accum_map (M.singleton k [e])) M.empty l++str2Numeric :: (Num a, Ord a) => String -> Maybe a+str2Numeric s = _str2Numeric s 0+ where+ _str2Numeric [] i = Just i+ _str2Numeric (sh:st) i | isDigit sh = let nexti = (i*10) + (fromIntegral $ digitToInt sh)+ in case i > nexti of+ False -> _str2Numeric st nexti+ True -> Nothing+ | otherwise = Nothing++truncLiterary :: String -> Int -> String+truncLiterary str n = case length str > n of+ True -> take (n - 3) str ++ "..."+ False -> str++readKeyPress :: IO ()+readKeyPress = do+ buf <- malloc :: IO (Ptr Word32)+ hWaitForInput stdin (negate 1)+ hGetBufNonBlocking stdin buf 4+ free buf++pressAnyKey :: IO ()+pressAnyKey = putStrLn "" << readKeyPress << putStr "\n-------Press any key to continue...-------"++bool2num :: Num a => Bool -> a+bool2num False = 0+bool2num True = 1++whenJust :: Maybe (IO ()) -> IO ()+whenJust mb_ioa = whenJustA () mb_ioa++whenJustA :: a -> Maybe (IO a) -> IO a+whenJustA a mb_ioa =+ case mb_ioa of+ Nothing -> return a+ Just ioa -> ioa++------------------------------------------+-- STM++modifyTVar :: TVar a -> (a -> (a, b)) -> STM b+modifyTVar tv f = do+ v <- readTVar tv+ let (a, b) = f v+ writeTVar tv a+ return b++modifyTVar_ :: TVar a -> (a -> a) -> STM ()+modifyTVar_ tv f = do+ v <- readTVar tv+ writeTVar tv (f v)++tryReadTChan :: TChan a -> STM (Maybe a)+tryReadTChan tch = do+ emp <- isEmptyTChan tch+ case emp of+ True -> return Nothing+ False -> Just `liftM` readTChan tch++-----------------------------+-- few public helpers++-- data System.IO.IOMode = ReadMode | WriteMode | AppendMode | ReadWriteMode+dump :: Show a => a -> Int -> IOMode -> IO ()+dump a i m = do+ h <- openFile ("./dump.out" ++ show i ++ ".hs") m+ hPutStr h $ show a+ hClose h++invDump :: Show a => Int -> IOMode -> a -> a+invDump i m a = unsafePerformIO $ do+ dump a i m+ return a++watch :: Show a => a -> a+watch a = unsafePerformIO $ do+ putStrLn_paged 22 $ show a+ return a++watchCond :: Show a => Bool -> a -> a+watchCond cond = if cond then watch else id++putStrLn_paged :: Int -> String -> IO ()+putStrLn_paged page_size s = f $ lines s+ where+ f lines_list =+ let (to_print, to_next_itera) = splitAt page_size lines_list+ in do putStrLn (concat $ intersperse "\n" to_print)+ case null to_next_itera of+ True -> return ()+ False -> f to_next_itera << hGetChar stdin << putStrLn "\n-------Press any key to continue...-------"++traceShowPaged :: Show a => Int -> a -> b -> b+traceShowPaged n a b = unsafePerformIO $ do+ putStrLn_paged n $ show a+ return b++traceCond :: Show a => Bool -> Int -> a -> b -> b+traceCond cond n a b = if cond then traceShowPaged n a b else b
+ PriorityChansConverger.cabal view
@@ -0,0 +1,56 @@+Name: PriorityChansConverger+Synopsis: + Read single output from an array of inputs - channels with priorities+Description: + Abbreviation for the @PriorityChansConverger@ is PCC.+ .+ Based on @STM.TChan@, extended with capacity control. + Wrapped into IO version is also available. + .+ When user reads from the PCC, the choice is made - from which channel + to read. System selects a nonempty channel, whose (CurrentPriority, + StartPriority) tuple is max. The side effect of the channel selection + is it's /CurrentPriority/ decrease by one, if it's value becomes + less than one, then the /CurrentPriority/ is set to /StartPriority/.+ .+ For the main API interface:+ .+ @import Control.Concurrent.PriorityChansConverger@+ .+ The realization probably isn't + very fast and isn't good at memory economy, since it uses fresh high + level primitive - STM. It wasn't intended to be used + with millions of channels. But it's max throughput comparing to the + ordinary @Chan@ throughput is to be estimated (will do it in some + future version).+Version: 0.1+category: Concurrency+Author: Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>+Maintainer: Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru>+Copyright: Copyright (c) 2009-2010 Andrejs Sisojevs+License: LGPL+License-file: COPYRIGHT+Stability: experimental+Tested-with: GHC ==6.10.4+Cabal-version: >=1.6+Build-type: Simple+Extra-source-files: + COPYING + COPYRIGHT + doinst.sh+ examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs+Build-depends: base ==4.*, containers -any, stm -any+Exposed-modules: + Control.Concurrent.ConcurrentUISupport+ Control.Concurrent.STM.TChan.ExtsCommons+ Control.Concurrent.STM.TChan.TChanL+ Control.Concurrent.STM.TChan.TChanL_+ Control.Concurrent.STM.TChan.TChanB+ Control.Concurrent.STM.TChan.TChanB_+ Control.Concurrent.PriorityChansConverger+ Control.Concurrent.PriorityChansConverger.Commons+ Control.Concurrent.PriorityChansConverger.PriorityChansConverger+ Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM+Other-modules: + Data.MyHelpers+ Data.MyHelpers_
+ Setup.hs view
@@ -0,0 +1,4 @@+#!/usr/bin/env runghc++import Distribution.Simple+main = defaultMain
+ doinst.sh view
@@ -0,0 +1,8 @@+#! /bin/sh+clear++sudo runghc Setup unregister+sudo runghc Setup configure --user+sudo runghc Setup build+sudo runghc Setup haddock+sudo runghc Setup install
+ examples/SimplePCCTestEnvironment/SimplePCCTestEnvironment.hs view
@@ -0,0 +1,442 @@+{- +Copyright (C) 2009-2010 Andrejs Sisojevs <andrejs.sisojevs@nextmail.ru> + +All rights reserved. + +For license and copyright information, see the file COPYRIGHT +-} + +-------------------------------------------------------------------------- +-------------------------------------------------------------------------- + +{- +abbr. PCC = PriorityChansConverger + +Table of content: + +1. Let's define system structure first. +1.1. Sources of input to our PCC. +1.2. Output of our PCC reader. +1.3. Put together PCC, it's input sources and it's output reader. + +2. Now, when the infrastructure is ready let's define some test cases and run them. +2.1. Test cases set. Not a complete set. If anybody wants to test it seriously, that would be great. +2.2. Running tests. + +3. Representations (instances of Show class). + +Appendix: Some helper functions. + +------------------------------------------- +Direct PCC usage is to be found only in parts 1.1., 1.2. and 1.3. + +-} + +{-# LANGUAGE ScopedTypeVariables, BangPatterns #-} + +module SimplePCCTestEnvironment where + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.ConcurrentUISupport -- this module is provided by PCC package +import Control.Concurrent.STM +import Control.Concurrent.PriorityChansConverger -- this module is provided by PCC package +import Control.Monad +import qualified Control.Exception as E +import Data.List +import qualified Data.Map as M +import Data.Map (Map) +import qualified Safe.Failure as SF +import System.IO +import System.IO.Unsafe + +__useDumps = True + +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 1. Let's define system structure first. +---------------------------------------------------------------------- +-- 1.1. Sources of input to our PCC. + +type PCCFillerID = Int -- PCC requires anything, that is of class Ord +type PCCFillerState = Int +type PCCFillerOutput = PCCFillerState +type PCCFillerCloser = ThreadFinisher +type MicrosecondsForDelay = Int +type SecondsForDelay = Int +data PCCFiller = + PCCFiller { + pccfInputID :: PCCFillerID -- local type + , pccfStartPriority :: StartPriority -- Int + , pccfChanCapacity :: ChanCapacity -- Int + , pccfPermitToBlock :: PermitToBlock -- Bool + , pccfDelayBetweenIterations :: MicrosecondsForDelay + , pccfStateMV :: MVar PCCFillerState + , pccfStateTransition :: PCCFillerState -> PCCFillerState + , pccfStopFiller :: MVar PCCFillerCloser + } + +defaultPCCFiller :: IO PCCFiller +defaultPCCFiller = do + state_mv <- newMVar 1 + stopper <- newMVar $ return () + return PCCFiller { + pccfInputID = undefined + , pccfStartPriority = undefined + , pccfChanCapacity = undefined + , pccfPermitToBlock = undefined + , pccfDelayBetweenIterations = undefined + , pccfStateMV = state_mv + , pccfStateTransition = (+ 1) + , pccfStopFiller = stopper + } + +runPCCFiller :: PCCFiller -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCFiller pccf pcc cuis = do + when __useDumps $ dump "" ("PCCFiller" ++ show channelID) WriteMode + interrupter_tv <- atomically $ newTVar False + cuisWrite cuis ("PCC filler (ID: " ++ show channelID ++ ") started.") + modifyMVar_ + (pccfStopFiller pccf) + (\ _ -> return $ do + cuisWrite cuis ("Signal to finish PCC filler (ID: " ++ show channelID ++ ") is sent.") + atomically $ modifyTVar_ interrupter_tv (const True) + ) + _ <- forkIO $ threadCycle interrupter_tv + return () + where + channelID = pccfInputID pccf + ------------ + threadCycle :: TVar InterruptShouldWe -> IO () + threadCycle interrupter_tv = _cycle + where + _cycle :: IO () -- It's possible to make this thread more STMish, use Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM directly. + _cycle = do + stop <- atomically $ readTVar interrupter_tv + case stop of + True -> do + cuisWrite cuis ("PCC filler (ID: " ++ show channelID ++ ") finished.") + return () -- exit + False -> do + state <- modifyMVar + (pccfStateMV pccf) + (\ current_state -> + return (pccfStateTransition pccf current_state, current_state) + ) + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump1" $ dump2 (return "\n----Before-write-----------------------------\n") ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump2" $ dump2 (showPCC pcc) ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode -- ######################################### + br_or_mb_failure <- case pccfPermitToBlock pccf of + True -> stmInterruptableWriteInPCC (readTVar interrupter_tv) channelID state pcc + False -> Right `liftM` writeInPCC False channelID state pcc + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump3" $ dump2 (return "\n----After-write------------------------------\n") ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCFiller:dump4" $ dump2 (showPCC pcc) ("PCCFiller" ++ show (pccfInputID pccf)) AppendMode -- ######################################### + case br_or_mb_failure of + Left () -> cuisWrite cuis ("Writing to channel (ID: " ++ show channelID ++ ") was interrupted." ) + Right Nothing -> cuisWrite cuis ("Wrote to channel (ID: " ++ show channelID ++ "): " ++ show (channelID, state) ) + Right (Just BadKey_FRWPCC) -> cuisWrite cuis ("Write to channel (ID: " ++ show channelID ++ ") FAILED: bad key." ) + Right (Just ChanFull_FRWPCC) -> cuisWrite cuis ("Write to channel (ID: " ++ show channelID ++ ") FAILED: not enough space in channel." ) + stop2 <- atomically $ readTVar interrupter_tv + unless stop2 $ threadDelay (pccfDelayBetweenIterations pccf) + _cycle + +runPCCFillersArray :: [PCCFiller] -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCFillersArray pccf_list pcc cuis = mapM_ (\ pccf -> runPCCFiller pccf pcc cuis) pccf_list + +stopPCCFiller :: PCCFiller -> IO () +stopPCCFiller = join . readMVar . pccfStopFiller + +stopPCCFillersArray :: [PCCFiller] -> IO () +stopPCCFillersArray = mapM_ stopPCCFiller + +---------------------------------------------------------------------- +-- 1.2. Output of our PCC reader. +---------------------------------------------------------------------- + +type PCCReaderID = Int + +data PCCReader = + PCCReader { + pccrReaderID :: PCCReaderID + , pccrPermitToBlock :: PermitToBlock + , pccrMicrosecondsForDelay :: MicrosecondsForDelay + , pccrStopFiller :: MVar PCCFillerCloser + } + +defaultPCCReader :: IO PCCReader +defaultPCCReader = do + stopper <- newMVar $ return () + return PCCReader { + pccrReaderID = undefined + , pccrPermitToBlock = undefined + , pccrMicrosecondsForDelay = undefined + , pccrStopFiller = stopper + } + +runPCCReader :: PCCReader -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCReader pccr pcc cuis = do + when __useDumps $ dump "" f_name WriteMode + interrupter_tv <- atomically $ newTVar False + cuisWrite cuis ("PCC reader (ID: " ++ show (pccrReaderID pccr) ++ ") started.") + _ <- forkIO $ threadCycle interrupter_tv + modifyMVar_ + (pccrStopFiller pccr) + (\ _ -> return $ do + cuisWrite cuis ("Signal to finish PCC reader (ID: " ++ show (pccrReaderID pccr) ++ ") is sent.") + atomically $ modifyTVar_ interrupter_tv (const True) + ) + return () + where + f_name = "PCCReader" ++ show (pccrReaderID pccr) + ------------ + threadCycle :: TVar InterruptShouldWe -> IO () + threadCycle interrupter_tv = _cycle + where + _cycle :: IO () -- It's possible to make this thread more STMish, use Control.Concurrent.PriorityChansConverger.PriorityChansConvergerSTM directly. + _cycle = do + stop <- atomically $ readTVar interrupter_tv + case stop of + True -> do + cuisWrite cuis ("PCC reader (ID: " ++ show (pccrReaderID pccr) ++ ") finished.") + return () -- exit + False -> do + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump1" $ dump2 (return "\n----Before-read------------------------------\n") f_name AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump2" $ dump2 (showPCC pcc) f_name AppendMode -- ######################################### + br_or_elem <- case pccrPermitToBlock pccr of + True -> stmInterruptableReadFromPCC (readTVar interrupter_tv) pcc + False -> do + mb_r <- readFromPCC False pcc + return $ case mb_r of + Nothing -> Left () + Just r -> Right r + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump3" $ dump2 (return "\n----After-read-------------------------------\n") f_name AppendMode + when __useDumps $ reportExceptionIfAny_2 cuis "runPCCReader:dump4" $ dump2 (showPCC pcc) f_name AppendMode -- ######################################### + case (pccrPermitToBlock pccr, br_or_elem) of + (True , Left ()) -> cuisWrite cuis ("Read from PCC (reader ID: " ++ show (pccrReaderID pccr) ++ ") is interrupred.") + (False, Left ()) -> cuisWrite cuis ("Read from PCC (reader ID: " ++ show (pccrReaderID pccr) ++ ") failed: PCC is empty.") + (_ , Right (k, e)) -> cuisWrite cuis ("Read from PCC (reader ID: " ++ show (pccrReaderID pccr) ++ "): " ++ show (k, e)) + stop2 <- atomically $ readTVar interrupter_tv + unless stop2 $ threadDelay (pccrMicrosecondsForDelay pccr) + _cycle + +stopPCCReader :: PCCReader -> IO () +stopPCCReader = join . readMVar . pccrStopFiller + +runPCCReadersArray :: [PCCReader] -> PriorityChansConverger PCCFillerID PCCFillerOutput -> ConcurrentUISupport -> IO () +runPCCReadersArray pccr_list pcc cuis = mapM_ (\ pccr -> runPCCReader pccr pcc cuis) pccr_list + +stopPCCReadersArray :: [PCCReader] -> IO () +stopPCCReadersArray = mapM_ stopPCCReader + +---------------------------------------------------------------------- +-- 1.3. Put together PCC, it's input sources and it's output reader. +---------------------------------------------------------------------- + +listOfFillers_to_ChansMapToConstructPCC :: [PCCFiller] -> Map PCCFillerID (StartPriority, ChanCapacity) +listOfFillers_to_ChansMapToConstructPCC pccf_list = + M.fromList $ map + (\ pccf -> + ( pccfInputID pccf + , (pccfStartPriority pccf + , pccfChanCapacity pccf + ) ) + ) + pccf_list + +constructPCC :: [PCCFiller] -> ConcurrentUISupport -> IO (PriorityChansConverger PCCFillerID PCCFillerOutput) +constructPCC pccf_list cuis = newPriorityChansConverger_wCUIS (Just cuis) $ listOfFillers_to_ChansMapToConstructPCC pccf_list +-- When constructing new PCC, it is possible to use 'newPriorityChansConverger' +-- instead of 'newPriorityChansConverger_wCUIS' which isn't using +-- any 'ConcurrentUISupport' + +runFillersAndPCCAndReaders :: [PCCFiller] -> [PCCReader] -> ConcurrentUISupport -> IO ThreadFinisher +runFillersAndPCCAndReaders pccf_list pccr_list cuis = do + pcc <- constructPCC pccf_list cuis + runPCCFillersArray pccf_list pcc cuis + runPCCReadersArray pccr_list pcc cuis + interrupter_tv <- atomically $ newTVar False + _ <- forkIO $ do + atomically $ do + stop <- readTVar interrupter_tv + case stop of + True -> return () + False -> retry + stopPCCFillersArray pccf_list + stopPCCReadersArray pccr_list + return (atomically $ writeTVar interrupter_tv True) + +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 2. Now, when the infrastructure is ready let's define +-- some test cases and run them +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 2.1. Test cases set. +-- Not a complete set. If anybody wants to test it seriously, +-- that would be great. + +type InputMode = Int +inputChansSet :: InputMode -> IO [PCCFiller] +inputChansSet 1 = sequence [] +inputChansSet 2 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 0, pccfPermitToBlock = False, pccfDelayBetweenIterations = 1000000 } ] +inputChansSet 3 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 1, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } ] +inputChansSet 4 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 5, pccfChanCapacity = 3, pccfPermitToBlock = False, pccfDelayBetweenIterations = 900000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 2, pccfStartPriority = 3, pccfChanCapacity = 5, pccfPermitToBlock = False, pccfDelayBetweenIterations = 1200000 } + ] +inputChansSet 5 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 4, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 2, pccfStartPriority = 8, pccfChanCapacity = 6, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 3, pccfStartPriority = 6, pccfChanCapacity = 8, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 4, pccfStartPriority = 4, pccfChanCapacity = 10, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + ] +inputChansSet 6 = sequence [ defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 1, pccfStartPriority = 10, pccfChanCapacity = 4, pccfPermitToBlock = True, pccfDelayBetweenIterations = 200000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 2, pccfStartPriority = 8, pccfChanCapacity = 6, pccfPermitToBlock = False, pccfDelayBetweenIterations = 200000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 3, pccfStartPriority = 6, pccfChanCapacity = 8, pccfPermitToBlock = False, pccfDelayBetweenIterations = 1000000 } + , defaultPCCFiller >>= \ pccf -> return pccf { pccfInputID = 4, pccfStartPriority = 4, pccfChanCapacity = 10, pccfPermitToBlock = True, pccfDelayBetweenIterations = 1000000 } + ] +__inputChansSet_lowerBound = 1 +__inputChansSet_higherBound = 6 + +type OutputMode = Int +readerSet :: OutputMode -> IO [PCCReader] +readerSet 1 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 1000000 } ] +readerSet 2 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 1000000 } ] +readerSet 3 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 500000 } ] +readerSet 4 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 500000 } ] +readerSet 5 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 1500000 } ] +readerSet 6 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 1500000 } ] +readerSet 7 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 500000 } + , defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 2, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 500000 } + ] +readerSet 8 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 200000 } + , defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 2, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 1500000 } + ] +readerSet 9 = sequence [ defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 1, pccrPermitToBlock = True, pccrMicrosecondsForDelay = 150000 } + , defaultPCCReader >>= \ pccr -> return pccr { pccrReaderID = 2, pccrPermitToBlock = False, pccrMicrosecondsForDelay = 200000 } + ] +__readerConfig_lowerBound = 1 +__readerConfig_higherBound = 9 + +---------------------------------------------------------------------- +-- 2.2. Running tests. +---------------------------------------------------------------------- + +type RunTest = Bool +runTest :: (RunTest, InputMode, OutputMode, SecondsForDelay) -> ConcurrentUISupport -> IO () +runTest (test_shouldwe, input_mode, output_mode, run_time) cuis = + E.handle + (\ se -> print (se :: E.SomeException)) + $ do pccr_list <- readerSet output_mode + pccf_list <- inputChansSet input_mode + case test_shouldwe of + False -> + let str = "" + ++ "PCCFillers(" ++ show (length pccf_list) ++ "): \n" + ++ (join $ intersperse "\n" $ map show pccf_list) + ++ "\n-----------------------------------\n" + ++ "PCCReaders(" ++ show (length pccf_list) ++ "): \n" + ++ (join $ intersperse "\n" $ map show pccr_list) + in cuisWrite cuis str + True -> do + finisher <- runFillersAndPCCAndReaders pccf_list pccr_list cuis + threadDelay (run_time * 1000000) + finisher + +main = do + cuis <- defaultConcurrentUISupportIO_logInFile "PCCTest.log" WriteMode -- provided by the PCC package, Control.Concurrent.SimpleUI + runConcurrentUISupport cuis + cuisWrite cuis ("Welcome to SimplePCCTestEnvironment v0.1!\n ") + _cycle cuis + cuisFinish cuis + where + _cycle :: ConcurrentUISupport -> IO () + _cycle cuis = do + cuisWrite cuis ("Please input mode (True/False, inputChansSet #{" ++ show __inputChansSet_lowerBound ++ ".." ++ show __inputChansSet_higherBound ++ "}, readerSet #{" ++ show __readerConfig_lowerBound ++ ".." ++ show __readerConfig_higherBound ++ "}, time (in seconds) for how long to run test). ") + cuisWrite cuis ("Or type ':q' for exit.") + cuisWrite cuis ("If first is True, then test is run, otherwise chosen inputChansSet and readerSet are shown.") + cuisWrite cuis ("Second parameter should be index of inputChansSet, third - of readerSet.") + cuisWrite cuis ("Example inputs: (True, 1, 1, 20) or (False,1,2,10), but not (1,2,3,4) nor (True," ++ show (__inputChansSet_lowerBound - 1) ++ "," ++ show (__readerConfig_higherBound + 1) ++ ",-1) nor (True,2,3,4,5)") + cuisWrite cuis ("Input > ") + mode <- cuisReadLn cuis "main thread" -- (True, 5, 9, 20) -- (True, 6, 6, 30) + case mode == ":q" of + True -> return () + False -> do + err_or_mode <- (E.try $ SF.read mode) :: IO (Either E.SomeException (RunTest, InputMode, OutputMode, SecondsForDelay)) + case err_or_mode of + Left se -> cuisWrite cuis (show se) >> cuisNewPage cuis + Right tm@(test_shouldwe, input_mode, output_mode, run_time) -> + let cond_1 = input_mode `betweenI` (__inputChansSet_lowerBound, __inputChansSet_higherBound) + cond_2 = output_mode `betweenI` ( __readerConfig_lowerBound, __readerConfig_higherBound) + cond_3 = run_time > 0 + cond = cond_1 && cond_2 && cond_3 + in case cond of + False -> return () + True -> runTest tm cuis >> threadDelay 1000000 >> cuisNewPage cuis + _cycle cuis + +---------------------------------------------------------------------- +---------------------------------------------------------------------- +-- 3. Representations (instances of Show class). +---------------------------------------------------------------------- +---------------------------------------------------------------------- + +instance Show PCCFiller where + show pccf = + "PCCFiller {" + ++ "\n ID = " ++ (show $ pccfInputID pccf) + ++ "\n StartPriority = " ++ (show $ pccfInputID pccf) + ++ "\n ChanCapacity = " ++ (show $ pccfChanCapacity pccf) + ++ "\n PermitToBlock = " ++ (show $ pccfPermitToBlock pccf) + ++ "\n MicrosecondsForDelay = " ++ (show $ pccfDelayBetweenIterations pccf) + ++ "\n PCCFillerState = " ++ (show $ unsafePerformIO $ readMVar $ pccfStateMV pccf) + ++ "\n}" + +instance Show PCCReader where + show pccr = + "PCCReader {" + ++ "\n ID = " ++ (show $ pccrReaderID pccr) + ++ "\n MicrosecondsForDelay = " ++ (show $ pccrMicrosecondsForDelay pccr) + ++ "\n PermitToBlock = " ++ (show $ pccrPermitToBlock pccr) + ++ "}" + +---------------------------------------------------------------------- +-- Appendix: Some helper functions. +---------------------------------------------------------------------- + +infixr 1 << +(<<) :: Monad m => m b -> m a -> m b +f << x = x >> f + +betweenI :: Ord a => a -> (a, a) -> Bool +betweenNI :: Ord a => a -> (a, a) -> Bool +betweenI a (l,h) = a >= l && a <= h +betweenNI a (l,h) = a > l && a < h + +infixr 1 `countIn` +countIn :: Eq a => a -> [a] -> Int +countIn a list = _count list 0 + where + _count [] !i = i + _count (h:t) !i = + case h == a of + True -> _count t (i + 1) + False -> _count t i + +modifyTVar_ :: TVar a -> (a -> a) -> STM () +modifyTVar_ tv f = do + v <- readTVar tv + writeTVar tv (f v) + +-- data System.IO.IOMode = ReadMode | WriteMode | AppendMode | ReadWriteMode +type WhatToDump = String +type FileSuffix = String +dump :: WhatToDump -> FileSuffix -> IOMode -> IO () +dump a s m = do + h <- openFile ("./dump.out." ++ s ++ ".hs") m + hPutStr h $ show a + hClose h + +dump2 :: IO WhatToDump -> FileSuffix -> IOMode -> IO () +dump2 a s m = do + h <- openFile ("./dump.out." ++ s ++ ".hs") m + a >>= hPutStr h + hClose h + --a >>= putStrLn